channel 与队列族

本章导读

第 28 章结尾预告了"另一条路线"，本章兑现。回顾一下 28.7 用 cond 做生产者消费者要凑齐的三件套：一个队列容器存数据、一把 mutex 保护它、一个 cond 传达"有货了/有位了"的通知，外加 while 谓词循环防虚假唤醒——每件都对，拼起来才对一个，漏一件就是事故。channel<T> 把这三件事收进一个类型：它自带缓冲、自带互斥、自带等待唤醒，send/recvInto 两个调用就是全部协作协议。如果说第 28 章教你"怎么造锁"，本章就是给你一把"已经造好的锁"——而且锁的钥匙孔设计得让你很难插错。本章依次讲：创建与容量（实测发现容量向上取整到 2 的幂）、六个收发操作、阻塞与关闭的完整语义、七种合法元素类型、与 thread 的经典配合模式，然后是标准库队列族四种（它们比 channel 更轻，但各司其职），最后用一个三级流水线实战收官，并复盘一次真实踩出来的流水线死锁。

29.1 从锁到通道

并发编程有句老话：不要靠共享内存来通信，要靠通信来共享内存。第 28 章走的是前半句的反路——所有线程盯着同一块内存，用锁维持秩序。这条路没有错（28 章的数素数实战就是这么赢的），但它把"哪些数据共享、哪段代码持锁"的分析负担全压在你肩上，28.4 的三张竞态面孔就是分析失误的代价。

channel 走后半句：数据不再共享，而是从一个线程手里复制一份、递交到另一个线程手里。发送方送出去之后不再拥有那份消息，接收方拿到的是自己的副本——竞态的温床（共享可变状态）在结构上就不存在了。协作协议也简化成一句话：有货就收，没货就等；关门就走。

代价同样真实：消息要复制（29.6 与深入细节 2），吞吐受缓冲容量与调度影响。所以本章后半的队列族提供了"我知道我的拓扑，给我最轻的管子"的选项——单生产单消费就不必付多生产者竞争的同步成本。选型见 29.9。

29.2 创建与容量

声明类型 channel<T>，构造用泛型调用 channel::<T>(容量)：

channel<int> ch = channel::<int>(2)    // 显式容量
channel<int> def = channel::<int>()     // 省略容量：默认 64 槽
auto autoCh = channel::<int>(1)          // auto 推断同样可行

容量的真实语义是本章第一个实测发现：容量会被向上取整到 2 的幂，最小为 2；传 0 或省略等价于默认 64。逐档实测（用 trySend 打满计数）：

为什么取整到 2 的幂？因为底层是无锁环形缓冲，下标回绕用位与代替取余（深入细节 1）。写代码时不要依赖精确容量——把容量当"缓冲量级"而非"精确上限"：想要紧背压就给小容量（1 也至少有 2 槽），想要削峰就给几十上百。

容量参数必须是整数，构造函数最多接收一个容量参数，违反均在编译期拦下（报错原文见常见错误）。

29.3 收发六操作

channel 的全部收发 API 是五个 bool 方法加一个超时变体，搭配两个观察方法：

channel<int> ch = channel::<int>(2)

// 发送：阻塞版与非阻塞版
bool ok1 = ch.send(42)        // 满则等待空间；已关闭返回 false
bool ok2 = ch.trySend(43)     // 满或已关闭立即返回 false

// 接收：目标必须是可以写入的具名变量，类型与元素类型完全一致
int v = 0
bool got1 = ch.recvInto(v)        // 空则阻塞等待；关闭且排空返回 false
bool got2 = ch.tryRecvInto(v)     // 空立即返回 false

// 超时接收：ms < 0 等价阻塞，ms = 0 立即检查，ms > 0 限时等待
bool got3 = ch.recvTimeoutInto(v, 100)   // 最多等 100ms

// 观察
println(ch.count())      // 当前缓冲消息数（瞬时值）
println(ch.isClosed())   // 是否已关闭

三个要点实测确认。第一，recvTimeoutInto 超时不碰目标变量：v 预置 999，空队列超时返回 false 后 v 仍是 999——可以安全复用接收变量。第二，ms 三态：-1 等价 recvInto 阻塞（实测另一线程 100ms 后送值成功接到），0 等价 tryRecvInto 立即检查，正数限时。第三，count() 是瞬时值：多线程下刚读完就可能过时，只能用于观测与调试，不能当同步依据——"count() > 0 所以 recvInto 不会阻塞"这种推理在并发下不成立。

所有接收变体的第一个参数都必须是具名、可写、类型精确匹配的变量：const 变量、闭包捕获变量、类型不符都会被编译期拦下（坑 3、坑 4）。这个限制是有意的——通道那头是另一个线程，写入目标必须是一块生命周期明确、独占可写的存储。

29.4 阻塞实录

"阻塞"不是形容词，是线程真的停下来等。两组实测为证。第一组：容量 2 的 channel 先填满，主线程再 send 第三个——它会卡在 send 里，直到 300ms 后消费者醒来取走一个腾出空间：

[main] filled, count=2
[main] send(3) begin (queue full)
[consumer] woke up, start recv      // 300ms 后消费者醒来
[consumer] got 1                    // 取走一个，腾出空间
[main] send(3) returned true        // 主线程的 send 此刻才返回

如果 send 不阻塞，"send(3) returned"会立刻出现在 begin 之后。实测顺序证明主线程老老实实等了消费者一手。

第二组：消费者在空 channel 上 recvInto 干等，主线程 100ms 后直接 close——close 会唤醒所有阻塞中的接收者，recvInto 返回 false，线程得以收拾退出（exitCode -1 是我们约定的"被关门"标记）：

channel<int> ch = channel::<int>(1)
auto waiter = func() as int
    int v = 0
    if ch.recvInto(v) then   // 空队列：阻塞在此
        return v
    end
    return -1                // 被 close 唤醒走到这里
end
thread t = thread(waiter)
time.sleep(100)
ch.close()
println(t.waitTimeout(2000))   // true：线程被唤醒并退出
println(t.exitCode())          // -1

这组"关门唤醒"语义是 channel 优雅退出的地基：消费者不需要轮询、不需要退出标志位，while ch.recvInto(v) 一个循环就是完整的生命周期（29.7）。

29.5 关闭语义

close 之后的 channel 行为规则实测完整，四条全部可以直接依赖：

这套"关而不丢"的语义正是广播式终止信号：生产者关门的瞬间，它等于向所有消费者宣布"货就这些了，收完下班"——已在队列里的货照样送达，消费者收完最后一条自然拿到 false 退出。

另外两条生命周期规则：channel 是 move-only 资源句柄（与 thread、mutex 同族），不能聚合进容器或当普通值复制；离开作用域时自动关闭队列并释放全部未消费消息——忘记 close 不会泄漏句柄，但等待中的消费者也就永远等不到那个 false 了，所以"谁关门"仍是设计的一部分（29.7 给出标准答案）。

29.6 元素类型

channel<T> 当前允许七种元素：bool、整数、浮点、string、time、point、var——共同点是都具备稳定的消息复制语义。实测全家福：

channel<string> cs = channel::<string>(2)
string src = "hello"
cs.send(src)
src = "changed"           // 发送后修改原变量
string got = ""
cs.recvInto(got)
println(got)              // hello：队中消息不受影响

channel<var> cv = channel::<var>(4)
cv.send(123)               // var 通道可以混装类型
cv.send("abc")
cv.send(1.25)
var x = null
cv.recvInto(x)
println(int(x))          // 123
cv.recvInto(x)
println(string(x))       // abc

复制语义是 channel 安全性的核心：普通值发送时复制进消息箱；string 发送时立即复制 UTF-8 字节、接收线程里重建；var 发送时提升为共享 xvalue、接收方接管新值。一句话——消息一旦出手，就与发送方再无瓜葛，这正是"靠通信共享内存"的落地方式（实现证据见深入细节 2）。

被拒之门外的是另一类：class、func、generator<T>、array<int> 等涉及复杂对象生命周期或 move-only 句柄的类型，语义阶段直接拒绝（坑 1）。class 是引用语义句柄（第 21、27 章），让它过通道等于重新引入共享——与 channel 的设计哲学相悖，所以干脆不许。想让两个线程共享一个 class 实例？那就回到第 28 章：传句柄走共享路线，自己配锁。

29.7 与 thread 的配合

channel 与 thread 是天作之合，两个模式覆盖八成场景。模式一，工厂闭包发 channel：channel 声明在顶层，工厂闭包捕获它（第 28 章的工厂模式原样适用）：

channel<int> ch = channel::<int>(8)
int producerDone = 0
mutex m = mutex()

func makeProducer(int base) as func() as int
    auto p = func() as int
        for i in range(1, 6)
            ch.send(base + i)
        end
        m.lock()
        producerDone += 1
        bool last = producerDone == 3
        m.unlock()
        if last then
            ch.close()              // 最后一个完成的生产者关门
        end
        return 0
    end
    return p
end

auto consumer = func() as int
    int v = 0
    int sum = 0
    while ch.recvInto(v)   // 模式二：关闭且排空后 false，循环自然结束
        sum += v
    end
    return sum
end

thread c = thread(consumer)
thread p1 = thread(makeProducer(0))
thread p2 = thread(makeProducer(100))
thread p3 = thread(makeProducer(200))
p1.wait()
p2.wait()
p3.wait()
c.wait()
println(c.exitCode())   // 1545：三个生产者 15 条消息一条不差

模式二，while recvInto 优雅退出：消费者的整个生命周期就是一个 while 循环，没有退出标志、没有轮询、没有 sleep 试探。这两个模式合起来的要点是"关门的人要可数"：单生产者直接关；多生产者用计数器认出自己的最后一个；生产者数量都不可数时，就由第三方（主线程 wait 完所有生产者后）关。

一个顺口提醒：for-in 遍历 channel 是不支持的（坑 5），while recvInto 就是官方的"遍历到关门"写法。

29.8 队列族四种

标准库还提供四个编译器识别的泛型原生队列：SpscQueue<T>（单生产单消费）、MpscQueue<T>（多生产单消费）、MpmcQueue<T>（多生产多消费）是有界并发队列，WaitQueue<T> 是支持阻塞与超时等待的等待队列。构造统一为 XxxQueue::<T>(容量)：

MpmcQueue<string> q = MpmcQueue::<string>(64)
q.tryPush("work")                // 非阻塞投递，满返回 false
string work = ""
q.tryPopInto(work)                 // 非阻塞领取，空返回 false

WaitQueue<int> waits = WaitQueue::<int>(64)
int v = 0
waits.popInto(v)                   // 阻塞领取：空则等待（仅 WaitQueue 有）
waits.popTimeoutInto(v, 1000)      // 限时等待 1000ms

方法集实测有一个与文档口径的出入要钉死：阻塞式的 popInto/popTimeoutInto 只存在于 WaitQueue，三种无锁队列只有非阻塞的 tryPush/tryPopInto，调用 popInto 会吃 [sem] unknown queue method。完整方法集对照：

其余语义实测：容量与 channel 一样向上取整到 2 的幂（1→2、3→4、5→8、100→128）；isDrained = 已关闭且已排空——关闭但队列里还有残留时 isDrained 是 false，残留可以照常 tryPopInto 取出，取空后才变 true；close 后 tryPush 返回 false；close 会唤醒 WaitQueue 上阻塞的 popInto（返回 false）。元素类型支持 bool/整数/浮点/string/time 等，string 同样是复制语义；var 元素当前有坑——声明 Queue<var> 运行期段错误（退出码 139，check 阶段不拦），请勿使用（已按编译器缺陷上报）。

并发正确性实测：SpscQueue 单生产单消费 5000 条，消费和 12502500 分毫不差；MpscQueue 三生产一消费 3000 条，和 31501500 精确；MpmcQueue 四生产两消费 8000 条、加锁汇总 128004000，与理论值完全一致、队列 leftover=0。拓扑选对，它们就是即插即用的无锁通道。

29.9 选型对照

五条路线摆在一起，按拓扑与需求挑：

经验法则三条。一，默认从 channel 起步：阻塞语义省掉自旋烧 CPU，close 广播省掉退出标志，代码最短。二，拓扑固定且性能敏感再上无锁队列：它们没有阻塞接收，消费者空转时得自己处理（自旋、sleep 或配合 sem）。三，需要"拿走消息时还要看一眼挑一挑"才回 28 章手工三件套——channel 与队列族的接口都是严格 FIFO，没有优先级与条件领取。

实战：三级流水线

把全章串起来：生产者线程发 20 个任务，三个工人从 jobs 领任务、平方后投进 results，主线程汇总。工人们在 jobs 关门后排空退出，最后一个退出的工人负责关 results——每个 channel 的关门人都"可数"，全程无一处显式加锁保护数据：

channel<int> jobs = channel::<int>(8)
channel<int> results = channel::<int>(8)
int workerDone = 0
mutex m = mutex()

auto producer = func() as int
    for i in range(1, 21)
        jobs.send(i)
    end
    jobs.close()                 // 发完即关门：通知工人退出
    return 20
end

func makeWorker() as func() as int
    auto w = func() as int
        int job = 0
        int processed = 0
        while jobs.recvInto(job)         // jobs 关闭排空后退出
            results.send(job * job)
            processed += 1
        end
        m.lock()
        workerDone += 1
        bool last = workerDone == 3
        m.unlock()
        if last then
            results.close()               // 最后一个工人关结果通道
        end
        return processed
    end
    return w
end

thread p = thread(producer)
thread w1 = thread(makeWorker())
thread w2 = thread(makeWorker())
thread w3 = thread(makeWorker())

int got = 0
int total = 0
int sumSq = 0
while results.recvInto(got)   // results 关闭排空后退出
    sumSq += got
    total += 1
end

p.wait()
w1.wait()
w2.wait()
w3.wait()
println("results=" + string(total) + " sumSq=" + string(sumSq))
// results=20 sumSq=2870（1²+2²+…+20² = 2870，分毫不差）
println("workers: " + string(w1.exitCode()) + "/" + string(w2.exitCode()) + "/" + string(w3.exitCode()))
// workers: 12/4/4（任务谁抢到算谁的，负载自然不均但总量守恒）

对比第 28 章 cond 版的同类程序：没有了队列容器、没有了谓词循环、没有了 signal 时机斟酌，锁只剩一把（给 workerDone 计数用）。但别把"结构简单"误解为"随便连都不会死"——这个流水线在写成"主线程先发完 jobs 再收 results"时真实死锁过一次，复盘见深入细节 3。

深入细节

⚠️ 常见错误

class Box
    int value = 0
end
channel<Box> ch = channel::<Box>(1)

channel<int> ch = channel::<int>(1)
bool ok = ch.send("bad")

channel<int> ch = channel::<int>(1)
float value = 0.0
bool ok = ch.recvInto(value)

channel<int> ch = channel::<int>(1)
bool ok = ch.wait()

channel<int> ch = channel::<int>(4)
for v in ch
    println(v)
end

本章小结

复习题

1. channel::<int>(3) 实际有多少个槽位？channel::<int>(100) 呢？为什么实现要做这种取整？
2. channel 关闭之后，send、缓冲未排空的 recvInto、缓冲已排空的 recvInto 各自什么表现？close 再调一次会怎样？
3. send 与 trySend、recvInto 与 tryRecvInto 的区别各是什么？什么场景下 trySend 返回 false 比 send 阻塞更有用？
4. recvTimeoutInto(v, ms) 的 ms 取负数、零、正数时分别是什么语义？超时返回 false 时 v 会被改动吗？
5. 为什么 channel 拒绝 class 元素却能接受 string？从消息复制语义角度解释。
6. 单生产单消费的固定管道，SpscQueue 与 channel 都能胜任，两者各付出什么、得到什么？

编程练习