我在找一个浏览器端JavaScript与 Java的 ExecutorService的等价实现。

要求：

• 使用Web Workers来实现真正的并行（不仅仅是Promise.all的并发）。
• 拥有固定数量的工作线程（类似于Executors.newFixedThreadPool(n)）。
• 允许在循环中提交大量任务。
• 稍后通过等待返回的Promise来收集结果。
• 尽管任务在并行执行，结果必须按提交顺序进行处理。

在Java中，我会写成类似下面这样：

ExecutorService executor = Executors.newFixedThreadPool(100);

List<Future<Result>> futures = new ArrayList<>();

for (Chunk chunk : chunks) {
  futures.add(executor.submit(() -> loadChunk(chunk)));
}

for (Future<Result> future : futures) {
  process(future.get());
}

executor.shutdown();

我想在浏览器端的JavaScript中，使用Web Workers实现类似的模式。

问题：

• 在浏览器端JavaScript中实现固定大小的工作池是否有标准或公认的做法？ 每个提交的任务应该新建一个Worker，还是应该复用Worker？
• 在并行执行任务的同时，如何保持提交顺序？
• 是否存在在Web Workers之上提供类似ExecutorService的抽象的现有库？

我只对浏览器端的JavaScript感兴趣（不涉及Node.js的 Worker线程）。

解决方案

用法

import { executor } from '@/async/Executor'
import GetParamsHistoryRunnable from '@/.../workers/GetParamsHistoryRunnable.ts?worker&url'

const paramIds = [1926735, 1926738, 1926741]
const period = 30 * 24 * 60 * 60 * 1000
const batch = 24 * 60 * 60 * 1000
const fromInclusive = Date.now() - period
const toExclusive = Date.now()

async function runClient() {
  const threshold = Date.now()

  const futures = []
  for (let from = fromInclusive; from < toExclusive; from += batch) {
    futures.push(executor.submit<ParameterHistoryUpdate[]>(GetParamsHistoryClientRunnable, [paramIds, from, Math.min(from + batch, toExclusive)]))
  }

  let paramsCount = 0
  for (const future of futures) {
    for (const param of await future) {
      paramsCount++
    }
  }

  console.log(`loaded ${paramsCount} params in ${Date.now() - threshold} ms`)
}

/GetParamsHistoryRunnable.ts

import { realtimeClient, type ParameterHistoryUpdate } from '@/app/.../realtimeClient'
import { StreamParamsHistoryRequestSchema, type StreamParamsHistoryRequest } from '@/app/.../realtimeapi/v1/realtimeapi_pb'

import { WorkerRunnable } from '@/async/WorkerRunnable'
import { create } from '@bufbuild/protobuf'

class GetParamsHistoryRunnable extends WorkerRunnable<[number[], number, number], ParameterHistoryUpdate[]> {
  protected async run(paramIds: number[], fromMs: number, toMs: number): Promise<ParameterHistoryUpdate[]> {
    const history: ParameterHistoryUpdate[] = []

    const req: StreamParamsHistoryRequest = create(StreamParamsHistoryRequestSchema, {
      paramIds: paramIds.map(id => BigInt(id)),
      fromInclusive: BigInt(Math.trunc(fromMs)),
      toExclusive: BigInt(Math.trunc(toMs)),
    })

    for await (const msg of realtimeClient.streamParamsHistory(req)) {
      history.push({
        id: Number(msg.paramId),
        value: msg.value,
        ts: Number(msg.timestamp),
        status: msg.status,
      })
    }

    return history
  }
}

new GetParamsHistoryRunnable()

/async/Executor.ts

type Progress = {
  completed: number
  queued: number
  running: number
}

type Task<T = unknown> = {
  id: number
  runnable: string
  args: unknown[]
  resolve: (value: T) => void
  reject: (reason?: unknown) => void
}

type WorkerRef = {
  worker: Worker
  runnable: string
  busy: boolean
}

export class Executor {
  private workers: WorkerRef[] = []
  private queue: Task[] = []
  private running = new Map<number, { task: Task; worker: WorkerRef }>()
  private taskCounter = 0
  private closed = false

  constructor(private workerCount: number, private onProgress: (progress: Progress) => void = () => { }) {
  }

  submit<T>(runnable: string | URL, args: unknown[] = []): Promise<T> {
    if (this.closed) return Promise.reject(new Error("Executor is closed"))
    return new Promise((resolve, reject) => {
      const task: Task<T> = { id: ++this.taskCounter, runnable: runnable.toString(), args, resolve, reject }
      this.queue.push(task as Task)
      this.runWorker()
    })
  }

  private runWorker() {
    while (this.queue.length > 0) {
      const task = this.queue[0]

      let workerRef = this.workers.find(worker => !worker.busy && worker.runnable === task.runnable)
      if (!workerRef && this.workers.length < this.workerCount) {
        workerRef = this.createWorker(task.runnable)
        this.workers.push(workerRef)
      }
      if (!workerRef)
        workerRef = this.workers.find(worker => !worker.busy)
      if (!workerRef)
        return
      if (workerRef.runnable !== task.runnable) {
        this.destroyWorker(workerRef)
        workerRef = this.createWorker(task.runnable)
        this.workers.push(workerRef)
      }

      this.queue.shift()
      this.running.set(task.id, { task, worker: workerRef })

      workerRef.busy = true
      workerRef.worker.postMessage({ id: task.id, args: task.args })
    }
  }

  private createWorker(runnable: string): WorkerRef {
    const worker = new Worker(runnable, { type: "module" })
    const workerRef: WorkerRef = { worker, runnable, busy: false }

    worker.onmessage = e => {
      workerRef.busy = false
      const { id, result, error } = e.data
      const running = this.running.get(id)
      if (!running) return

      this.running.delete(id)
      if (error)
        running.task.reject(error instanceof Error ? error : new Error(String(error)))
      else
        running.task.resolve(result)

      this.onProgress({ completed: id, queued: this.queue.length, running: this.running.size })
      this.runWorker()
    }

    worker.onerror = e => {
      workerRef.busy = false
      for (const [id, running] of this.running) {
        if (running.worker === workerRef) {
          this.running.delete(id)
          running.task.reject(e instanceof Error ? e : new Error(String(e)))
          break
        }
      }

      this.destroyWorker(workerRef)
      this.runWorker()
    }

    return workerRef
  }

  private destroyWorker(workerRef: WorkerRef) {
    workerRef.worker.terminate()
    const index = this.workers.indexOf(workerRef)
    if (index >= 0)
      this.workers.splice(index, 1)
  }

  close() {
    this.closed = true

    for (const task of this.queue)
      task.reject(new Error("Executor is closed"))

    for (const { task } of this.running.values())
      task.reject(new Error("Executor is closed"))

    for (const workerRef of this.workers)
      workerRef.worker.terminate()

    this.workers = []
    this.queue = []
    this.running.clear()
  }
}

export const executor = new Executor(Math.max(1, (navigator.hardwareConcurrency ?? 5) - 1))

/async/WorkerRunnable.ts

export abstract class WorkerRunnable<Args extends unknown[], Result> {
  constructor() {
    self.onmessage = async event => {
      const { id, args } = event.data

      try {
        const result = await this.run(...args as Args)
        self.postMessage({ id, result })
      } catch (e) {
        self.postMessage({ id, error: e instanceof Error ? e : new Error(String(e)) })
      }
    }
  }

  protected abstract run(...args: Args): Promise<Result> | Result
}