task 编排与生成器

本章导读

第 30 章结尾留了一个口子：批量任务我们用手写循环收集 array<future<T>> 再逐个 await——能用，但"等全部完成"这种集团军操作该有正式编制。本章上半补齐它：task.spawn 把任意普通函数直接后台化（连 async 声明都省），task.all 一组 future 等齐收数组，task.race/whenAny 先到先得，TaskGroup 给一组长寿命任务提供结构化编组与关闭。下半换主角：生成器。普通函数一口气 return 全部结果，生成器用 yield 一件一件地交——调用时不执行、迭代才推进、退出即关闭，惰性让它能表达"无穷数列"这种数组装不下的东西；加上 await 能力就是 async generator，于是"从异步来源逐件拉数据"也有了标准写法。实战把两半焊在一起：生成器供料、task.all 并发加工的批处理管道。读完本章，卷六四件武器——thread、channel、async/future、task/generator——就全部到手了。

31.1 从 future 到编排

第 30 章的 future 是"一张提货单"，本章的 task API 回答三个批量问题：

• 怎么把普通函数直接变成任务？——task.spawn：不用写 async 声明，任何签名完整的 func 值扔进后台线程，拿回 future。
• 怎么等一组任务？——task.all 等全部、task.race 抢第一、task.whenAny 抢第一并报告是谁。
• 怎么管理一组长寿命任务？——TaskGroup：可以加人、等人、开子组、关门，比散装 future 数组多一层生命周期。

它们与 future 的关系：task 函数生产与消费的都是 future<T>，30 章的 await/cancel/then/catch/finally 全部原样适用——编排不是新类型，是 future 之上的组合运算。

31.2 task.spawn

task.spawn(fn, 参数...) 把普通函数放到后台线程执行，返回 future。函数以 func 变量或 lambda 的形式给出，参数逐个跟在后面：

func delayed(int value, int delayMs) as int
    time.sleep(delayMs)
    return value
end

// 先存 func 值，再 spawn
func(int, int) as int job = delayed
future<int> f = task.spawn(job, 7, 50)
println(await f)                // 7

// lambda 直接传
future<int> g = task.spawn(func(int x) as int => x * 2, 21)
println(await g)                // 42

一个容易踩的门槛：裸函数名不行——task.spawn(delayed, 7, 50) 报 [sem] task.spawn first argument must be a func value with a complete signature，先赋给 func 变量即可（官方用例也是这个写法）。spawn 与手写 async func 的取舍：函数已经存在且签名合适时用 spawn 零改造；要 await 别的 future、要精细控制时才写 async func。spawn 产生的 future 与 async 产生的完全同族——cancel、then、入组 task.all 都可以。

31.3 task.all 等全部

task.all([f1, f2, ...]) 接收 future 数组（字面量或 array<future<T>>），等全部完成，返回 future<array<T>>——await 它得到按输入顺序排列的结果数组：

async func slow(int v, int ms) as int
    time.sleep(ms)
    return v
end

// 数组字面量输入；结果按输入顺序，与完成先后无关
array<int> all = await task.all([slow(1, 100), slow(2, 50), slow(3, 10)])
println(all[0])        // 1：最慢的那个仍在首位

// typed 数组输入；named group future 可重复 await
array<future<int>> jobs = array[slow(10, 20), slow(20, 10)]
future<array<int>> grp = task.all(jobs)
grp.wait()
array<int> a1 = await grp
array<int> a2 = await grp       // 重复 await，结果稳定

约束两条：数组非空、所有 future 的完成值类型一致（违反在编译期拦下，坑 4）。任何一个 source 失败，group 传播那个原始异常（慢的同伙不受影响，可以继续各自 await）。30.8 手写循环收割的模式，正式场合就换成 task.all——语义更明确，还自带"按输入顺序"的保证。

31.4 task.race 与 whenAny

这对兄弟都是"先到先得"，区别在给你什么：

int winner = await task.race([slow(1, 200), slow(2, 50), slow(3, 300)])
println(winner)               // 2：50ms 的最快

int idx = await task.whenAny([slow(30, 200), slow(40, 50), slow(50, 300)])
println(idx)                  // 1：第二路（索引从 0 起）

race 的陪跑者会被请求取消（协作式，与 30.6 同规则）——赢家的值到手后，loser 的后续 await 抛"future cancel requested."。whenAny 只报告索引不动 source，想取消谁你自己决定。参数约束与 task.all 相同：非空、完成值类型一致。

31.5 取消传播

取消一组任务有两条路，实测行为各不相同：

future<int> s1 = slow(10, 50)
future<int> s2 = slow(20, 400)
future<array<int>> grp = task.all([s1, s2])
task.cancel(s2)
try
    array<int> r = await grp
catch e
    println(e.message)      // future cancel requested.
end
println(await s1)           // 10：无辜者照常

设计直觉：task.all 承诺"给齐一组结果"，任何一票作废承诺就无法兑现，所以一票取消整组即废——但它不搞连坐，未取消的 source 该交付还交付。记住 task.cancel 只认 future/task：往里面塞 channel、thread、generator 都在编译期拒绝（[sem] task.cancel 参数必须是 future<T> 或 task<T>）。

31.6 TaskGroup

task.all/race 是"一次性合影"——组建立刻、成员固定。要"边跑边加人、统一管理、关门打烊"的长寿命编组，用 task.group()：

TaskGroup g = task.group()
println(g.valid)              // true
println(g.count)              // 0
println(g.add(123))           // false：只收 future

future<int> a = task.spawn(job, 10, 30)
future<int> b = task.spawn(job, 20, 60)
g.add(a)                         // true
g.add(b)                         // true，count=2
println(g.waitTimeout(2000))  // true：等全员完成，count 归零

// 子组：parent 可以孵化 child，各自编组
TaskGroup child = g.child()
child.add(task.spawn(job, 99, 20))
future<null> done = child.joinAsync()   // 异步等待全组
await done
println(child.reapCompleted()) // 1：收割已完成数

// 关门：close 后 add 返回 false
g.close()
println(g.add(a))              // false
println(g.joinTimeout(100))    // true：已关闭且清空，立即会合

方法/属性一览：valid 句柄有效、count 在组任务数、add(future) 加人（非 future/已关闭返回 false）、waitTimeout(ms) 限时等全员、child() 开子组、joinAsync() 返回 future<null> 供 await、joinTimeout(ms) 限时会合、reapCompleted() 收割已完成数、close() 关门。与 task.all 的分工：一次性并发批处理用 all；服务式的"任务来了就加、定期收割、收工关门"用 TaskGroup。

31.7 generator 与 yield

把函数返回类型写成 generator<T>，它就不再"一口气算完返回"，而是用 yield 一件一件交货；调用方用 for-in 逐件消费（支持带索引的双变量形态）：

func nums(int count) as generator<int>
    int i = 0
    while i < count
        i += 1
        yield i
    end
end

int total = 0
for v in nums(3)
    total += v
end
println(total)            // 6

for i, v in nums(3)
    println(string(i) + " -> " + string(v))   // 0->1 1->2 2->3
end

规则要点：yield 只能出现在返回 generator<T> 的函数体内（坑 1）；yield 的值类型必须匹配 T（坑 3）；函数里可以用裸 return 提前收尾（后续 yield 不再执行），但不能携带返回值（坑 2）；生成器内 throw 会在消费方的迭代处原样抛出。元素类型开放 bool/整数/浮点/string/time/point——string 与 point 本章实测可用（var、class、func、静态容器被拒，报错口径与 channel 同款）。

31.8 惰性与单次消费

生成器有两条铁律，都是实测钉死的。铁律一，惰性：调用时函数体一行都不执行，每次迭代推进到下一个 yield 停住，break 后剩余部分永远不再跑：

[main] before call
[main] after call, before iterate   // 调用已返回，函数体纹丝未动
[gen] body started                  // 第一次迭代才启动
[main] got 1
[gen] after yield 1
[main] got 2                        // break：after yield 2 与 body ending 永不出现
[main] after loop

与 async 的"调用即启动"恰好相反——同一个语言里两种求值时机，别记混：async 是先跑起来等你取，generator 是你拉一下它动一下。

铁律二，单次消费：for 循环退出即关闭——无论是耗尽还是 break。已关闭的生成器再迭代，抛 GeneratorError"iterate null generator"（不捕获则进程以退出码 1 终止）；裸声明未赋值的 generator 迭代同样抛这个错。没有"暂停后继续"的公开操作：想消费多少，在同一次 for 循环里决定。这条限制是刻意的（深入细节 3），写代码时的推论是：生成器随用随造，别囤。

31.9 状态与无限流

生成器函数体内的局部变量在两次 yield 之间原样保持——这正是它能表达"无穷数列"的原因：数组必须装得下才存在，生成器只记"下一个怎么算"：

func fib() as generator<int>
    int a = 0
    int b = 1
    while true              // 无限循环不可怕：惰性只在你拉的时候走一步
        yield a
        int next = a + b
        a = b
        b = next
    end
end

int n = 0
int sum = 0
for v in fib()
    print(string(v) + " ")    // 0 1 1 2 3 5 8 13 21 34
    sum += v
    n += 1
    if n == 10 then
        break               // 限量靠 break（同时关闭生成器）
    end
end
println(sum)              // 88

多个生成器实例状态各自独立（实测：fib() 开两个实例，一个只拉首件得 0，另一个拉到 8，互不干扰）。一个边界要知道：generator 是 move-only 句柄，不能作为参数传递——想写"过滤生成器"接收另一个生成器会吃 [sem] generator 参数当前只支持 ...；管道组合要写在一个函数体内（练习 2 演示）。

31.10 async generator

生成器体内要用 await（从异步来源逐件拉数据），就是 async generator。两种声明形态完全等价，消费端用 for await（也支持带索引）：

async func addAsync(int a, int b) as int
    time.sleep(20)
    return a + b
end

// 形态一：async func ... as generator<int>
async func streamA(int start) as generator<int>
    int first = await addAsync(start, 1)
    yield first
    yield await addAsync(first, 2)
end

// 形态二：func ... as async generator<int>
func streamB(int start) as async generator<int>
    int first = await addAsync(start, 10)
    yield first
    yield await addAsync(first, 20)
end

int totalA = 0
for await v in streamA(100)
    totalA += v
end
println(totalA)          // 204（101 + 103）

for await i, v in streamB(1000)
    println("B[" + string(i) + "]=" + string(v))
end
// B[0]=1010  B[1]=1030

实测要点：async generator 同样惰性（调用不执行，迭代才推进）；for await 要求对方是 generator<T>（拿数组喂它报 [sem] for await requires generator<T>），但消费端 for 与 for await 对 sync/async 两种生成器都兼容——写法自由，for await 更像一种"这里面有异步"的阅读提示。throw、return、单次消费规则与同步生成器一致。

31.11 生成器 vs 数组

"返回一组值"有两条路，选型对照：

经验法则：结果集小且要反复用、要按下标取——返回数组；数据量大、逐件处理、边算边用、或者根本没有"全部"这个概念（无限流、实时流）——生成器。生成器还是"关注点分离"的好工具：生产逻辑（怎么算出下一个）与消费逻辑（拿到每个做什么）各自独立演化，中间的接口只有 yield 出来的 T。

实战：批处理管道

把本章两半焊起来：生成器惰性供料，每拉出一批就 task 化并发加工，task.all 统一收割。8 批任务各含 50ms 模拟 I/O——串行要 400ms 起步，并发实测 0.143s 收工：

// 供料：惰性产生批次
func batches(int count) as generator<int>
    int i = 0
    while i < count
        i += 1
        yield i * 100     // 每批的加工基数
    end
end

// 加工：对 [base, base+100) 内奇数计数，sleep 模拟 I/O
async func process(int base) as int
    int count = 0
    for n in range(base, base + 100)
        if n % 2 == 1 then
            count += 1
        end
    end
    time.sleep(50)
    return count
end

// 管道：拉一批发一批，全部发出后统一收割
array<future<int>> jobs
for base in batches(8)
    jobs.append(process(base))
end
array<int> results = await task.all(jobs)

int total = 0
for i in range(0, results.length)
    total += results[i]
end
println(total)           // 400（8 批 × 50 个奇数）

这个结构的美妙之处在于三层各司其职：生成器决定"有什么活"（改供料逻辑不动加工）、async func 决定"怎么干"（改加工逻辑不动编排）、task.all 决定"怎么会合"（改汇总逻辑不动前两层）。卷六到此收官：thread 给你发动机，channel 给你传动轴，async/future 给你操控台，task 与 generator 给你编队与流水线——四轮齐了。

深入细节

⚠️ 常见错误

func bad() as int
    yield 1
end

func bad() as generator<int>
    yield 1
    return 99
end

func bad() as generator<int>
    yield 1.5
end

array<int> r = await task.all([])

channel<int> ch = channel::<int>(1)
task.cancel(ch)

本章小结

复习题

1. task.spawn 的第一个参数接受什么形态？为什么裸函数名会被拒？
2. task.all 的结果数组按什么顺序排列？named group future 能否重复 await？
3. task.race 与 task.whenAny 的 await 结果分别是什么？race 对陪跑者做什么？
4. 入组后取消其中一个 source，group 和其他 source 各自什么下场？
5. 调用一个生成器函数时函数体执行吗？for 循环 break 之后再迭代该生成器会怎样？
6. 生成器与返回数组在内存占用和重复消费上有什么根本差别？什么场景必须选生成器？

编程练习