/** * 按key合并重复读取的通用资源加载器,限制活跃并发与结果缓存数量,等待队列本身没有长度上限。 * 多个调用者共享一次load但可独立取消;最后一个订阅离开才撤销底层请求。适用于可按key复用的资源, * 不保存业务操作/请求回执,也不替代runtime可靠队列。concurrency/cacheSize由调用方提供有效范围。 */ /** 创建共享加载实例。默认并发6、最多缓存16个key,缓存按最近peek顺序淘汰。 * undefined用作未命中标记,因此load返回undefined不会作为可复用命中。此工厂未运行时校验数值配置。 */ export function createResourceLoader(load: (key: string, signal: AbortSignal) => Promise, concurrency = 6, cacheSize = 16) { // 每次request各有Promise与abort监听,取消一个调用者不能拒绝仍需要结果的其它调用者。 type Subscriber = { resolve: (value: T) => void; reject: (error: unknown) => void; cleanup: () => void }; // 一个key的一次执行;controller属于共享job,订阅者signal不直接当底层signal。 type Job = { key: string; controller: AbortController; subscribers: Set }; // cache保存完成值,jobs保存当前可加入的尝试,queue只保存尚未占并发槽的job。 const cache = new Map(); const jobs = new Map(); const queue: Job[] = []; let active = 0; /** 命中后移到Map尾部,更新近用顺序;只读缓存,不启动网络。 */ function peek(key: string) { const value = cache.get(key); if (value !== undefined) { cache.delete(key); cache.set(key, value); } return value; } /** 从队头取得有订阅的任务,占槽后异步load,最终归还槽并继续排队任务。 * 在途任务即使被abort也要等Promise真正结束才释放并发名额。 */ function pump() { while (active < concurrency && queue.length) { const job = queue.shift()!; if (!job.subscribers.size) continue; active++; // 发起前再查共享取消状态,队列任务可在进入微任务前失去全部订阅。 Promise.resolve().then(() => { if (job.controller.signal.aborted) throw new DOMException("Aborted", "AbortError"); return load(job.key, job.controller.signal); }).then((value) => { if (!job.controller.signal.aborted) { cache.set(job.key, value); // 超出缓存容量淘汰最久未触及key,不影响仍在运行的其它job。 while (cache.size > cacheSize) cache.delete(cache.keys().next().value!); for (const subscriber of job.subscribers) { subscriber.cleanup(); subscriber.resolve(value); } } }, (error) => { for (const subscriber of job.subscribers) { subscriber.cleanup(); subscriber.reject(error); } }).finally(() => { // 旧取消任务结束时,同key可能已有新job;按对象身份删除,不能误删替代任务。 if (jobs.get(job.key) === job) jobs.delete(job.key); active--; pump(); }); } } /** 已取消立即拒绝;缓存命中立即返回;否则加入同key的在途/排队job,必要时新建一次执行。 */ function request(key: string, signal?: AbortSignal): Promise { if (signal?.aborted) return Promise.reject(new DOMException("Aborted", "AbortError")); const cached = peek(key); if (cached !== undefined) return Promise.resolve(cached); let job = jobs.get(key); if (!job) { job = { key, controller: new AbortController(), subscribers: new Set() }; jobs.set(key, job); queue.push(job); } const current = job; return new Promise((resolve, reject) => { /** 只撤销本订阅并解除监听。最后一个离开时移除排队项/可加入索引,再abort底层执行。 */ const abort = () => { current.subscribers.delete(subscriber); subscriber.cleanup(); reject(new DOMException("Aborted", "AbortError")); if (!current.subscribers.size) { if (jobs.get(key) === current) jobs.delete(key); const index = queue.indexOf(current); if (index >= 0) queue.splice(index, 1); current.controller.abort(); } }; const subscriber: Subscriber = { resolve, reject, cleanup: () => signal?.removeEventListener("abort", abort) }; current.subscribers.add(subscriber); signal?.addEventListener("abort", abort, { once: true }); pump(); }); } return { request, peek }; }