返回列表协作式多任务与异步语法糖
详解迭代器协议、Generator暂停恢复机制、yield双向传值原理、async/await编译产物分析及异步迭代器应用
协作式多任务与异步语法糖
Generator 函数是 JavaScript 实现协作式多任务的基础设施。它允许函数执行暂停和恢复，yield 表达式充当暂停点。async/await 是基于 Promise 和 Generator 的语法糖，V8 引擎直接在底层优化了 async 函数的执行，使其性能远超手动实现的 Generator runner。理解 Generator 的状态机本质是掌握异步编程高级模式（如 Redux-Saga、CSP 通道）的前提。
目录
- Generator 的状态机模型
- yield 的双向通信
- Generator 的迭代协议
- yield* 委托与组合
- 手动实现 async/await
- async Generator 与流处理
- V8 中 async 函数的优化
- 实战案例
- 深度追问
- 总结表格
1. Generator 的状态机模型
Generator 函数在调用时不执行函数体，而是返回一个 Generator 对象：
function* counter() {
console.log('start');
let i = ;
() {
i++;
}
}
gen = ();
.(gen.());
.(gen.());
.(gen.());
0
while
true
yield
const
counter
console
log
next
console
log
next
console
log
next
内部状态转换
┌─────────┐ next() ┌──────────┐ return/throw ┌───────────┐
│suspended │──────────►│ executing │───────────────►│ completed │
│ start │ └─────┬────┘ └───────────┘
└─────────┘ │ yield
▲ ▼
│ ┌──────────────┐
└──────────────│ suspended │
next() │ yield │
└──────────────┘
function* lifecycle() {
console.log('Phase 1: before first yield');
const a = yield 'first';
console.log('Phase 2: received', a);
const b = yield 'second';
console.log('Phase 3: received', b);
return 'done';
}
const g = lifecycle();
console.log(g.next());
console.log(g.next('hello'));
console.log(g.next('world'));
console.log(g.next());
2. yield 的双向通信
function* calculator() {
let result = 0;
while (true) {
const { op, value } = yield result;
switch (op) {
case 'add': result += value; break;
case 'sub': result -= value; break;
case 'mul': result *= value; break;
case 'div': result /= value; break;
case 'reset': result = 0; break;
}
}
}
const calc = calculator();
calc.next();
console.log(calc.next({ op: 'add', value: 10 }));
console.log(calc.next({ op: 'mul', value: 3 }));
console.log(calc.next({ op: 'sub', value: 5 }));
throw 与 return
function* errorHandler() {
try {
const val = yield 'waiting';
console.log('Got:', val);
} catch (e) {
console.log('Caught:', e.message);
yield 'recovered';
} finally {
console.log('Cleanup');
}
}
const g = errorHandler();
console.log(g.next());
console.log(g.throw(new Error('Oops')));
console.log(g.return('forced'));
3. Generator 的迭代协议
function* range(start, end, step = 1) {
for (let i = start; i < end; i += step) {
yield i;
}
}
for (const n of range(0, 10, 2)) {
console.log(n);
}
const arr = [...range(1, 6)];
const [first, second, ...rest] = range(0, 100);
无限序列与惰性求值
function* naturals() {
let n = 1;
while (true) yield n++;
}
function* map(iterable, fn) {
for (const item of iterable) {
yield fn(item);
}
}
function* filter(iterable, predicate) {
for (const item of iterable) {
if (predicate(item)) yield item;
}
}
function* take(iterable, n) {
let count = 0;
for (const item of iterable) {
if (count++ >= n) return;
yield item;
}
}
const result = [...take(
map(
filter(naturals(), n => n % 2 === 0),
n => n * n
),
5
)];
console.log(result);
4. yield* 委托与组合
function* inner() {
yield 'a';
yield 'b';
return 'inner-done';
}
function* outer() {
yield 1;
const result = yield* inner();
console.log('inner returned:', result);
yield 2;
}
console.log([...outer()]);
树遍历
function* inorder(tree) {
if (tree) {
yield* inorder(tree.left);
yield tree.value;
yield* inorder(tree.right);
}
}
const bst = {
value: 4,
left: { value: 2, left: { value: 1, left: null, right: null }, right: { value: 3, left: null, right: null } },
right: { value: 6, left: { value: 5, left: null, right: null }, right: { value: 7, left: null, right: null } }
};
console.log([...inorder(bst)]);
5. 手动实现 async/await
function co(generatorFn) {
return function(...args) {
const gen = generatorFn.apply(this, args);
return new Promise((resolve, reject) => {
function step(method, value) {
let result;
try {
result = gen[method](value);
} catch (err) {
return reject(err);
}
if (result.done) {
return resolve(result.value);
}
Promise.resolve(result.value).then(
val => step('next', val),
err => step('throw', err)
);
}
step('next', undefined);
});
};
}
const fetchUser = co(function* (id) {
const response = yield fetch(`/api/user/${id}`);
const user = yield response.json();
const posts = yield fetch(`/api/posts?userId=${user.id}`);
return yield posts.json();
});
async function fetchUserAsync(id) {
const response = await fetch(`/api/user/${id}`);
const user = await response.json();
const posts = await fetch(`/api/posts?userId=${user.id}`);
return await posts.json();
}
6. async Generator 与流处理
async function* fetchPages(baseUrl) {
let page = 1;
let hasMore = true;
while (hasMore) {
const response = await fetch(`${baseUrl}?page=${page}`);
const data = await response.json();
yield* data.items;
hasMore = data.hasNext;
page++;
}
}
async function getAllItems() {
const items = [];
for await (const item of fetchPages('/api/products')) {
items.push(item);
if (items.length >= 1000) break;
}
return items;
}
ReadableStream 集成
async function* streamToAsyncGenerator(stream) {
const reader = stream.getReader();
try {
while (true) {
const { done, value } = await reader.read();
if (done) return;
yield value;
}
} finally {
reader.releaseLock();
}
}
async function processLargeFile(url) {
const response = await fetch(url);
const decoder = new TextDecoder();
let lineBuffer = '';
for await (const chunk of streamToAsyncGenerator(response.body)) {
lineBuffer += decoder.decode(chunk, { stream: true });
const lines = lineBuffer.split('\n');
lineBuffer = lines.pop();
for (const line of lines) {
await processLine(line);
}
}
if (lineBuffer) await processLine(lineBuffer);
}
7. V8 中 async 函数的优化
async function nativeAsync() {
return await Promise.resolve(42);
}
const generatorVersion = co(function* () {
return yield Promise.resolve(42);
});
await 的快速路径
async function fast() {
const p = Promise.resolve(1);
const v = await p;
return v;
}
async function slow() {
const thenable = { then(cb) { cb(1); } };
const v = await thenable;
return v;
}
8. 实战案例
实战案例 1：可取消的异步流
async function* createCancellableStream(source, signal) {
const iterator = source[Symbol.asyncIterator]();
try {
while (true) {
if (signal.aborted) {
throw new DOMException('Aborted', 'AbortError');
}
const result = await Promise.race([
iterator.next(),
new Promise((_, reject) => {
signal.addEventListener('abort', () => {
reject(new DOMException('Aborted', 'AbortError'));
}, { once: true });
})
]);
if (result.done) return;
yield result.value;
}
} finally {
await iterator.return?.();
}
}
const controller = new AbortController();
const stream = createCancellableStream(fetchPages('/api'), controller.signal);
setTimeout(() => controller.abort(), 5000);
for await (const item of stream) {
console.log(item);
}
实战案例 2：Redux-Saga 风格的副作用管理
function createSagaMiddleware() {
const effects = {
call: (fn, ...args) => ({ type: 'CALL', fn, args }),
put: (action) => ({ type: 'PUT', action }),
take: (pattern) => ({ type: 'TAKE', pattern }),
fork: (saga, ...args) => ({ type: 'FORK', saga, args }),
};
function runSaga(store, saga) {
const gen = saga(effects);
function step(value) {
const { done, value: effect } = gen.next(value);
if (done) return;
switch (effect.type) {
case 'CALL':
effect.fn(...effect.args).then(
result => step(result),
err => gen.throw(err)
);
break;
case 'PUT':
store.dispatch(effect.action);
step();
break;
case 'TAKE':
store.subscribe((action) => {
if (action.type === effect.pattern) {
step(action);
}
});
break;
}
}
step();
}
return { runSaga, effects };
}
实战案例 3：并发限制的异步迭代
async function* mapConcurrent(iterable, fn, concurrency = 5) {
const iterator = iterable[Symbol.asyncIterator]
? iterable[Symbol.asyncIterator]()
: iterable[Symbol.iterator]();
const pending = new Set();
let done = false;
const results = [];
let resolveWait;
async function fillPool() {
while (pending.size < concurrency && !done) {
const { value, done: iterDone } = await iterator.next();
if (iterDone) { done = true; break; }
const promise = fn(value).then(result => {
pending.delete(promise);
results.push(result);
resolveWait?.();
});
pending.add(promise);
}
}
await fillPool();
while (pending.size > 0 || results.length > 0) {
if (results.length > 0) {
yield results.shift();
await fillPool();
} else {
await new Promise(r => { resolveWait = r; });
}
}
}
const urls = Array.from({ length: 50 }, (_, i) => `/api/${i}`);
for await (const data of mapConcurrent(urls, url => fetch(url).then(r => r.json()), 5)) {
console.log(data);
}
9. 深度追问
Q1：Generator 函数的 this 绑定是什么？
Generator 函数内部的 this 遵循普通函数的绑定规则。但 Generator 对象的原型方法（next/throw/return）中的 this 是 Generator 对象本身。不能对 Generator 函数使用 new（它不是构造函数）。
Q2：为什么 for await...of 不能用于同步可迭代对象？
实际上可以！ES2020 规范允许 for await...of 接受同步可迭代对象，它会将每个值包装在 Promise.resolve() 中。但这种用法有性能开销，应避免在不必要时使用。
Q3：async Generator 的 return() 会怎样？
调用 async generator 的 return() 方法会触发 finally 块执行，然后返回 { value: arg, done: true }。如果 finally 块中有 yield，它们仍会产出值，直到 finally 完成。
10. 总结表格
| 特性 | Generator | Async Function | Async Generator |
|---|
| 返回值 | Generator 对象 | Promise | AsyncGenerator 对象 |
| 暂停关键字 | yield | await | yield / await |
| 消费方式 | for...of / next() | await | for await...of |
| 取消机制 | return() | AbortController | return() + signal |
| V8 优化 | 一般 | 高度优化 | 一般 |
| 使用场景 | 推荐方式 | 原因 |
|---|
| 异步 I/O | async/await | 简洁、V8 优化 |
| 无限序列 | Generator | 惰性求值 |
| 流式数据 | Async Generator | 异步 + 惰性 |
| 协程调度 | Generator | 精细控制暂停/恢复 |
| 状态机 | Generator | 天然状态机模型 |
| 副作用管理 | Generator (Saga) | 可测试性 |
