协程

概念概述

协程（Coroutines）指的是：协程是可被挂起与恢复的函数，让许多逻辑任务在同一线程上协作式共享，而非每个任务独占一个线程。

从更严格的定义看，A coroutine is a generalised subroutine that can pause at suspension points, yielding control to a scheduler and later resuming from the same logical position with its local state preserved. 该概念在工程实践中的价值，在于它能帮助团队用统一的语言描述一个具体的质量属性或设计约束，从而在评审与协作中减少歧义。

为什么需要它

They underpin async and await, generators, and Go and Kotlin concurrency models that multiplex many tasks onto few operating-system threads. 它与 Software Transactional Memory、Happens-Before Relationship、Thread-Local Storage 等概念经常一起出现，用于解决相近的工程问题。理解它的适用场景，有助于在多种备选方案之间做出有依据的取舍，而不是凭直觉选择技术方案。

常见误解

• coroutines run in parallel on multiple CPU cores by definition
• a suspended coroutine keeps an operating-system thread blocked

需要避免的反模式

• calling blocking code inside a coroutine and stalling the scheduler
• treating coroutine-local state as if it were thread-local

实践要点

1. When a coroutine awaits, it unwinds to its caller and the underlying thread is free to run other work, so thousands of tasks can be in flight without thousands of threads.
2. Parallelism is a separate concern achieved by running tasks on a pool of threads or processors, so concurrency and parallelism are not the same thing.
3. Because a coroutine may resume on a different thread, any mutable state shared across suspension points needs synchronisation just as in threaded code.

参考来源

https://en.wikipedia.org/wiki/Coroutine

代码示例

export async function pool(items, worker, concurrency) {
  if (concurrency < 1) throw new RangeError("concurrency must be >= 1");
  const results = new Array(items.length);
  let next = 0;
  async function run() {
    while (next < items.length) {
      const index = next++;
      results[index] = await worker(items[index], index);
    }
  }
  const runners = [];
  for (let i = 0; i < Math.min(concurrency, items.length); i++) {
    runners.push(run());
  }
  await Promise.all(runners);
  return results;
}

export function mapPool(items, mapper, concurrency = 4) {
  if (!Array.isArray(items)) throw new TypeError("items must be an array");
  if (typeof mapper !== "function") throw new TypeError("mapper must be a function");
  if (!Number.isInteger(concurrency) || concurrency < 1) {
    throw new RangeError("concurrency must be a positive integer");
  }

  const results = new Array(items.length);
  let nextIndex = 0;

  async function worker() {
    while (nextIndex < items.length) {
      const index = nextIndex;
      nextIndex += 1;
      results[index] = await mapper(items[index], index);
    }
  }

  const workerCount = Math.min(concurrency, items.length);
  const workers = Array.from({ length: workerCount }, () => worker());
  return Promise.all(workers).then(() => results);
}