并发编程（九）：BlockingQueue 与 Channel——语言层的原子性、可见性与有序性

目录

• 0. 从上一篇继续
• 1. Message Passing 的发展历史
  • 1.1 Dijkstra：通过共享缓冲区协作
  • 1.2 RC 4000：通过系统消息接口通信
  • 1.3 Actor Model：通过异步消息驱动行为
  • 1.4 CSP：通过同步通信协调执行
  • 1.5 四种协作方式有什么区别？
  • 1.6 回到 Message Passing
• 2. 三种语言的消息交接规则分别保证什么？
  • 2.1 Java：BlockingQueue
    • 2.1.1 Java 有哪些 Queue？
    • 2.1.2 BlockingQueue 的两条并发规则
    • 2.1.3 SPSC：单 Producer、单 Consumer
    • 2.1.4 MPSC：多 Producer、单 Consumer
    • 2.1.5 MPMC：多 Producer、多 Consumer
    • 2.1.6 阻塞等待与 FIFO
  • 2.2 Go：Channel
    • 2.2.1 Go 有哪些 Channel？
    • 2.2.2 Channel 的通信与同步规则
    • 2.2.3 SPSC：单 Producer、单 Consumer
    • 2.2.4 MPSC：多 Producer、单 Consumer
    • 2.2.5 MPMC：多 Producer、多 Consumer
    • 2.2.6 Buffered / Unbuffered 与 FIFO
  • 2.3 CPython：queue.Queue
    • 2.3.1 Python 有哪些 Queue？
    • 2.3.2 Queue 的线程安全与同步语义
    • 2.3.3 SPSC：单 Producer、单 Consumer
    • 2.3.4 MPSC：多 Producer、单 Consumer
    • 2.3.5 MPMC：多 Producer、多 Consumer
    • 2.3.6 阻塞等待、FIFO 与任务完成
• 3. 下一篇：消息交接是怎么实现的？

0. 从上一篇继续

在第零篇中，我们从多个并发执行单元如何交换数据、协调工作的角度，提出了两种典型的并发协作模型：

• Shared Memory：多个执行单元访问同一份共享状态，通过同步机制协调访问。
• Message Passing：多个执行单元通过发送和接收消息交换数据、协调工作。

前面几篇主要沿着 Shared Memory 展开，从 Mutex、Atomic 到 ReadWriteLock，解释了这些同步机制如何提供 Atomicity、Visibility 和 Ordering。

这一篇转向Message Passing。在进入具体语言之前，我们先回顾这一并发协作思想的发展历史。

1. Message Passing 的发展历史

1.1 Dijkstra：通过共享缓冲区协作

1968 年，Dijkstra 在《Cooperating Sequential Processes》中研究顺序进程的协作，并使用一个 Producer、一个 Consumer 的例子：双方通过共享缓冲区交换数据，依靠 Semaphore 协调执行。

当缓冲区为空时，Consumer 需要等待；如果使用有界缓冲区，缓冲区满时 Producer 也需要等待。

1.2 RC 4000：通过系统消息接口通信

1969 年，Per Brinch Hansen 为 RC 4000 计算机设计了RC 4000 Multiprogramming System（多道程序操作系统），通过系统内核提供进程间消息通信机制。随后，他在 1970 年发表的《The Nucleus of a Multiprogramming System》中介绍了这一设计。

与 Dijkstra 的共享缓冲区方案不同，RC 4000 的系统核心（Nucleus）提供了明确的消息通信操作：

• send message：向指定进程发送消息。
• wait message：等待并接收消息。
• send answer：向消息发送方返回回复。
• wait answer：等待对应的回复。

系统内核负责分配消息缓冲区，并将消息放入指定接收方的队列。

把这四个操作放到一个 Producer、一个 Consumer 的例子里：

Producer 通过send message发送任务，Consumer 调用wait message取出任务；处理后，Consumer 可以通过send answer回复，Producer 再通过wait answer获取结果。

如果 Consumer 尚未准备接收，只要系统有可用的消息缓冲资源，Producer 就可以先完成发送并继续执行。

1.3 Actor Model：通过异步消息驱动行为

1973 年，Carl Hewitt、Peter Bishop 和 Richard Steiger 在《A Universal Modular ACTOR Formalism for Artificial Intelligence》中提出 Actor Model：Actor 拥有自身状态，通过异步消息与其他 Actor 协作，收到消息后按自身行为处理。

以 Producer–Consumer 为例，可以让 Buffer Actor 接收 Producer 的任务，再转交 Consumer：

这里的 Buffer Actor 和 Mailbox 只是示例设计，并非 Actor Model 要求的固定结构。

1.4 CSP：通过同步通信协调执行

1978 年，Tony Hoare 在《Communicating Sequential Processes》中提出了 CSP。

CSP 将程序组织成多个顺序执行的 Process，通过明确的输入与输出操作进行通信。其典型通信方式是同步会合（Rendezvous）：发送与对应的接收操作必须配对，通信才能完成。这里的 Process 是计算模型中的执行单元，不要求对应独立的操作系统进程。

回到 Producer–Consumer，如果 Producer 要把任务交给 Consumer，两者必须在通信操作上配对：

图中的 Rendezvous 是一次逻辑通信事件，并不代表中央缓冲区。如果 Consumer 尚未准备接收，Producer 就需要等待；双方配对后，任务才完成交接。

1.5 四种协作方式有什么区别？

需要注意，这四者并不是相互替代的，而是在不同背景下探索执行单元之间的通信与协作：Dijkstra 讨论进程协作与同步，RC 4000 提供操作系统消息机制，而 Actor Model 和 CSP 是更加完整的并发计算模型。

1.6 回到 Message Passing

Message Passing 的核心，是让执行单元通过显式的消息发送与接收交换数据、协调执行。。 通信可以有缓冲，也可以无缓冲：有缓冲时，消息可以先暂存，发送方与接收方异步交接；无缓冲时，消息不能暂存，必须等待收发双方就绪后同步交接。

接下来，我们分别从 JavaBlockingQueue、GoChannel和 CPythonqueue.Queue的语言规则出发，回答以下三个问题：

• Atomicity：怎样保证单次消息交接的正确性？
• Visibility：发送消息前准备的数据，接收方能否正确观察？
• Ordering：发送与接收之间，能够建立怎样的同步顺序？

2. 三种语言的消息交接规则分别保证什么？

2.1 Java：BlockingQueue

2.1.1 Java 有哪些 Queue？

Java 的Queue是一种存放待处理元素的集合接口。根据是否支持并发访问、是否需要等待，我们可以把常见实现分成三类：

其中，BlockingQueue有几种代表实现：

这里选用最典型的实现ArrayBlockingQueue来讲解。

最常用的两个操作是：

• put(e)：放入元素；队列满时等待可用空间。
• take()：取出并移除元素；队列空时等待新元素。

2.1.2 BlockingQueue 的两条并发规则

JDK 25 BlockingQueue 文档 规定：

“All queuing methods achieve their effects atomically.”

这条规则对应Atomicity。每次入队或出队都是原子的，不会因并发竞争而破坏队列状态。

同一文档的 Memory Consistency Effects 还规定：

一个线程将对象放入BlockingQueue之前的操作，happens-before 另一个线程随后访问或移除该元素之后的操作。

这条规则同时对应Visibility和Ordering。

对于 Visibility，Producer 在发送消息前完成的写入，对取得该消息的 Consumer 可见。

对于 Ordering，发送前的操作、消息交接以及接收后的操作，通过 happens-before 的传递性建立跨线程内存顺序。

继续沿用第零篇的count++，从 SPSC 扩展到 MPSC、MPMC：

classIncrement{intdelta;}BlockingQueue<Increment>queue=newArrayBlockingQueue<>(2);

2.1.3 SPSC：单 Producer、单 Consumer

// Producer 线程Incrementmessage=newIncrement();message.delta=1;queue.put(message);// Consumer 线程intcount=0;Incrementreceived=queue.take();count+=received.delta;// count == 1

对应的执行关系：

Producer Consumer message.delta = 1 │ │ program order ▼ queue.put(message) ─── 消息交接 ────────► queue.take() 返回 message │ │ program order ▼ read received.delta == 1 count = 1

put()成功放入一条消息，take()成功取出并移除它。这符合 Atomicity 的单次操作保证，但此时还没有多个 Producer 或多个 Consumer 之间的竞争。

delta是普通字段，没有使用volatile或 Atomic。假设消息入队后没有其他线程继续修改它，Producer 的字段写入 happens-before Consumer 的读取，所以 Consumer 能看到1，也符合 Visibility 和 Ordering 的保证。

2.1.4 MPSC：多 Producer、单 Consumer

// Producer A、B 各自在自己的线程执行一次Incrementmessage=newIncrement();message.delta=1;queue.put(message);// 唯一的 Consumer 线程intcount=0;for(inti=0;i<2;i++){count+=queue.take().delta;}// 两条消息均成功提交并被处理后，count == 2

一种可能的执行顺序：

Producer A Producer B Consumer put(A: +1) put(B: +1) take() -> +1 count: 0 -> 1 take() -> +1 count: 1 -> 2

Atomicity：两个 Producer 同时入队，也不会互相覆盖已成功提交的消息。只要两次put()均成功，Consumer 就能取出两条消息。

Visibility / Ordering：对 Producer A、B 各自发送的消息，前述 happens-before 规则分别成立；但两次并发put()谁先成功，并没有预定顺序。

这里count += delta并非原子指令，只是count始终由同一个 Consumer 修改，因此不会发生两个 Consumer 之间的丢失更新。

2.1.5 MPMC：多 Producer、多 Consumer

继续增加一个 Consumer。两个 Producer 仍然分别发送+1；现在由 Consumer A、B 并发调用take()：

// Producer A、B：各发送一个独立的 Increment 对象Incrementmessage=newIncrement();message.delta=1;queue.put(message);// Consumer A、B：各执行一次Incrementreceived=queue.take();intdelta=received.delta;

例如，两条消息分别由两个 Consumer 取走：

Producer A Producer B Consumer A Consumer B put(A: +1) put(B: +1) take() -> A take() -> B

Atomicity：同一条消息只会被一次成功的take()移除，不会被两个 Consumer 重复取走。图中 A、B 只是区分两次发送的消息对象，并不表示固定分配给哪个 Consumer。

Visibility / Ordering：无论哪个 Consumer 取得消息，都能依赖该消息对应 Producer 的入队前写入与接收后读取之间的 happens-before 关系。但消息出队的先后，并不保证两个 Consumer 业务处理完成的先后。

这里还存在一个边界：如果两个 Consumer 取得消息后仍直接修改同一个普通count：

// 反例：两个 Consumer 共享同一个 Counter 对象classCounter{intcount;}CountersharedCounter=newCounter();// 先创建，交给两个 Consumer 共用// Consumer A、B 各在自己的线程中执行一次intdelta=queue.take().delta;sharedCounter.count+=delta;

两条消息虽然各自被安全取出，但count仍可能因并发更新而只得到1。BlockingQueue 保证的是消息交接的原子性，而不是任意后续业务操作的原子性。

2.1.6 阻塞等待与 FIFO

以上三个场景还共享ArrayBlockingQueue的两个具体行为：

• 阻塞等待：队列为空时，take()等待；容量为2的队列满时，put()等待可用空间。
• FIFO：元素按照实际入队的顺序出队。但多个 Producer 的竞争入队顺序不预先确定，也不能据此推断多个 Consumer 的业务完成顺序。

2.2 Go：Channel

与 Java 不同，Go 没有提供一组与BlockingQueue对应的标准并发队列实现，而是直接在语言层提供了Channel，用于 Goroutine 之间的数据通信。

2.2.1 Go 有哪些 Channel？

根据 Go Language Specification，Channel 可以按照缓冲容量分为两种：

Channel 还可以限制通信方向：

方向限制与是否有缓冲是两个独立的维度。例如，一个 Buffered Channel 也可以通过chan<- T类型交给只负责发送消息的函数。

这里选用典型的Buffered Channel来讲解。

初始化一个容量为2的 Buffered Channel：

typeIncrementstruct{deltaint}jobs:=make(chanIncrement,2)

其中两个最重要的操作是：

• ch <- value：发送消息；缓冲区满时等待可用空间。
• value := <-ch：接收消息；缓冲区空时等待新消息。

2.2.2 Channel 的通信与同步规则

Go Language Specification 规定，同一个 Channel 可以被任意数量的 Goroutine 发送、接收，无需额外同步。

这条规则对应Atomicity。并发 Send / Receive 可以安全交接消息：每次成功的接收对应一条发送的消息，不会让多个 Consumer 重复取得同一条消息。
Go Memory Model — Channel Communication 还规定：

“A send on a channel is synchronized before the completion of the corresponding receive from that channel.”

这条规则同时对应Visibility和Ordering。

对于 Visibility，Producer 在发送消息前完成的写入，对完成对应接收的 Consumer 可见。

对于 Ordering，发送前的操作、Channel 的 Send / Receive，以及接收后的操作，通过 synchronized-before 和程序顺序建立 happens-before 关系。

下面继续用第零篇的count++，从 SPSC 扩展到 MPSC、MPMC。

2.2.3 SPSC：单 Producer、单 Consumer

// Producer Goroutinemessage:=Increment{delta:1}jobs<-message// Consumer Goroutinecount:=0received:=<-jobs count+=received.delta// count == 1

对应的执行关系：

Producer Consumer message.delta = 1 │ │ sequenced-before ▼ jobs <- message ─── 消息交接 ───────────► received := <-jobs 完成 │ │ sequenced-before ▼ read received.delta == 1 count = 1

一次成功的 Send 与 Receive 完成消息交接。此时还没有多个 Producer 或 Consumer 之间的竞争。

delta是普通字段。Go Channel 会传递Increment的值；发送前的写入 synchronized-before 对应接收完成，因此 Consumer 能读取到1，满足 Visibility 和 Ordering。

2.2.4 MPSC：多 Producer、单 Consumer

// Producer A、B 各自在自己的 Goroutine 执行一次jobs<-Increment{delta:1}// 唯一的 Consumer Goroutinecount:=0fori:=0;i<2;i++{received:=<-jobs count+=received.delta}// 两条消息均成功发送并被处理后，count == 2

一种可能的执行顺序：

Producer A Producer B Consumer send(A: +1) send(B: +1) receive -> +1 count: 0 -> 1 receive -> +1 count: 1 -> 2

Atomicity：两个 Producer 并发 Send 不会互相覆盖已成功发送的消息；两次发送各对应一次接收。

Visibility / Ordering：每条消息的发送与对应接收之间分别建立同步关系；两个 Producer 谁先发送成功并没有预定顺序。

2.2.5 MPMC：多 Producer、多 Consumer

// Producer A、B 各自在自己的 Goroutine 执行一次jobs<-Increment{delta:1}// Consumer A、B 各自在自己的 Goroutine 执行一次received:=<-jobs delta:=received.delta

例如，两条消息分别被两个 Consumer 接收：

Producer A Producer B Consumer A Consumer B send(A: +1) send(B: +1) receive -> A receive -> B

Atomicity：同一次发送的消息只由一个对应的 Receive 取得，不会被两个 Consumer 重复接收。图中的 A、B 仅用来区分消息，不代表固定的消费者分配规则。

Visibility / Ordering：每个 Consumer 都能依赖所接收消息的 Send / Receive 同步关系，但接收消息的顺序并不保证处理完成的顺序。

如果两个 Consumer 接收后仍修改同一个共享count：

// 反例：两个 Consumer 共享同一个普通计数器received:=<-jobs sharedCounter.count+=received.delta

即使两个 Receive 都正确取得消息，sharedCounter.count仍然存在 Data Race，可能发生丢失更新。Channel 保障消息交接，不会自动保护接收后的共享状态。

2.2.6 Buffered / Unbuffered 与 FIFO

前面的例子使用容量为2的 Buffered Channel：

• 等待：缓冲区空时 Receive 等待；缓冲区满时 Send 等待。
• FIFO：同一个 Channel 按消息实际发送的顺序交接，但并发 Producer 的发送先后并不预先确定，也不保证多个 Consumer 的业务完成顺序。

Go 还支持 Unbuffered Channel：

jobs:=make(chanIncrement)

它不存储消息，Send 和 Receive 必须配对，Send 才能完成。Go Memory Model 还规定：Unbuffered Channel 的对应 Receive synchronized-before Send 完成。这项额外保证不适用于一般的 Buffered Channel 发送完成时刻。

Channel 如何让等待中的 Goroutine 挂起，并在条件满足时恢复执行，是下一篇要回答的问题。

2.3 CPython：queue.Queue

与 Go 不同，Python 没有在语言语法层提供 Channel，而是通过标准库提供不同执行模型下的队列机制。

2.3.1 Python 有哪些 Queue？

Python 为不同的执行单元提供了不同的队列：

其中，queue提供线程安全队列；asyncio.Queue面向异步 Task，不提供线程安全保证；multiprocessing.Queue面向多进程通信，传递的对象需要经过序列化。

这里讨论 CPython 的线程，因此重点介绍标准库queue模块。

根据 Python queue 官方文档，它提供以下几种代表实现：

任务完成跟踪指通过task_done()标记任务处理完成，通过join()等待所有入队任务处理完成。

注意：Python 的 maxsize <= 0 表示缓冲区容量不设上限；Go 的 Channel 容量为 0 则表示没有缓冲区。

这里选用典型的实现queue.Queue来讲解。其中两个最重要的操作是：

两者默认阻塞等待，也可以通过block=False立即尝试，或通过timeout限制等待时间；未能放入或取出元素时，分别抛出queue.Full或queue.Empty。

2.3.2 Queue 的线程安全与同步语义

Python queue 官方文档 规定：

“TheQueueclass in this module implements all the required locking semantics.”

这条契约对应Atomicity：queue.Queue支持多个线程并发执行put()和get()，使用内部锁保证队列操作的线程安全；同一条消息不会被两个 Consumer 重复取出。

对于Visibility和Ordering，Producer 在put()之前准备消息，Consumer 通过get()取得该消息，再读取其字段。在 CPython 中，这一交接依靠queue.Queue内部的锁与条件变量完成线程同步，使 Consumer 能读取交接前准备好的数据。

。

继续沿用第零篇的count++，从 SPSC 扩展到 MPSC、MPMC：

fromdataclassesimportdataclassfromqueueimportQueue@dataclassclassIncrement:delta:intjobs=Queue(maxsize=2)

2.3.3 SPSC：单 Producer、单 Consumer

Producer 发送一条+1消息，唯一的 Consumer 接收后更新count：

# Producer Threadmessage=Increment(delta=1)jobs.put(message)# Consumer Threadcount=0received=jobs.get()count+=received.delta# count == 1

对应的执行关系：

Producer Consumer message.delta = 1 │ ▼ jobs.put(message) ─── Queue 内部同步 ───► jobs.get() 返回 message │ ▼ read received.delta == 1 count = 1

put()成功放入一条消息，get()取出并移除它。只有一个 Producer 和一个 Consumer 时，还没有同侧的并发竞争。

delta是普通字段。假设入队后不再有线程并发修改这个消息，Consumer 通过 Queue 完成交接后可以读取到1；这一可见性来自 Queue 的同步，而不是 GIL 自动保护了所有共享数据。

与第零篇通过ready协调读取不同，这里get()会在队列为空时等待消息，无需额外共享ready标记。

2.3.4 MPSC：多 Producer、单 Consumer

增加一个 Producer。两个 Producer 各发送一条+1消息，唯一的 Consumer 顺序累加：

# Producer A、B 各自在自己的 Thread 执行一次jobs.put(Increment(delta=1))# 唯一的 Consumer Threadcount=0for_inrange(2):received=jobs.get()count+=received.delta# 两条消息均成功提交并被处理后，count == 2

一种可能的执行顺序：

Producer A Producer B Consumer put(A: +1) put(B: +1) get() -> +1 count: 0 -> 1 get() -> +1 count: 1 -> 2

Atomicity：两个 Producer 并发调用put()不会覆盖彼此已成功入队的消息；单个 Consumer 可以依次取出两条消息。

Visibility / Ordering：每条消息都通过 Queue 完成交接与同步；两个 Producer 谁先成功入队没有预定顺序。

count += received.delta是普通复合操作，但这里仅由单个 Consumer 修改count，不涉及多个 Consumer 对同一计数器的竞争。

2.3.5 MPMC：多 Producer、多 Consumer

再增加一个 Consumer。两个 Producer 各提交一条+1，两个 Consumer 并发调用get()：

# Producer A、B 各自在自己的 Thread 执行一次jobs.put(Increment(delta=1))# Consumer A、B 各自在自己的 Thread 执行一次received=jobs.get()delta=received.delta

例如，两条消息分别由两个 Consumer 取得：

Producer A Producer B Consumer A Consumer B put(A: +1) put(B: +1) get() -> A get() -> B

Atomicity：成功的get()会从队列移除一个元素；同一条消息不会被两个 Consumer 同时取出。图中的 A、B 仅用于区分消息，不表示预先指定哪个 Consumer 接收。

Visibility / Ordering：取得消息的 Consumer 可以依赖 Queue 的线程同步读取入队前准备的数据；但出队顺序不等于两个 Consumer 的业务完成顺序。

如果两个 Consumer 之后仍然修改同一个共享计数器：

# 反例：两个 Consumer 共用 countercounter={"value":0}# 在创建两个 Consumer 之前共享# Consumer A、B 各自执行一次received=jobs.get()counter["value"]+=received.delta

队列安全取出两条消息，并不能保证两次counter["value"] += 1不发生丢失更新。不能依赖 GIL 把这个复合操作当作 Python 的原子性保证，Free-threaded CPython 更不能这样假设。

2.3.6 阻塞等待、FIFO 与任务完成

前面的三个场景还共享queue.Queue的具体行为：

• 阻塞等待：容量为2的队列满时，put()默认等待；空队列上的get()默认等待。maxsize=0表示不限制容量，而不是 Go 式无缓冲通信。
• FIFO：queue.Queue按实际入队顺序取出元素；并发 Producer 谁先入队无法预先确定，也不保证多个 Consumer 按出队顺序完成处理。
• 任务完成：get()只表示取出消息，task_done()用于报告任务处理完成，join()等待未完成任务计数归零。取出消息和处理完成不是一回事。

Queue 底层怎样通过 Lock 与 Condition 协调等待线程，是下一篇讨论的问题。

3. 下一篇：消息交接是怎么实现的？

这一篇停在语言层：JavaBlockingQueue和 GoChannel通过明确的通信规则保证消息交接及其内存顺序；CPythonqueue.Queue则依靠标准库的线程同步机制完成安全交接。

下一篇会把这些规则转换成实现层需要解决的问题，分别沿着 JavaArrayBlockingQueue、GoChannel和 CPythonqueue.Queue，从同步原语、Runtime 一直看到 OS 与 CPU。

本文首发于 ThinkerQAQ 的个人博客，由作者本人同步发布。原文可能持续修订，最新版本请以个人博客为准。