目录0. 从上一篇继续1. Message Passing 的发展历史1.1 Dijkstra通过共享缓冲区协作1.2 RC 4000通过系统消息接口通信1.3 Actor Model通过异步消息驱动行为1.4 CSP通过同步通信协调执行1.5 四种协作方式有什么区别1.6 回到 Message Passing2. 三种语言的消息交接规则分别保证什么2.1 JavaBlockingQueue2.1.1 Java 有哪些 Queue2.1.2 BlockingQueue 的两条并发规则2.1.3 SPSC单 Producer、单 Consumer2.1.4 MPSC多 Producer、单 Consumer2.1.5 MPMC多 Producer、多 Consumer2.1.6 阻塞等待与 FIFO2.2 GoChannel2.2.1 Go 有哪些 Channel2.2.2 Channel 的通信与同步规则2.2.3 SPSC单 Producer、单 Consumer2.2.4 MPSC多 Producer、单 Consumer2.2.5 MPMC多 Producer、多 Consumer2.2.6 Buffered / Unbuffered 与 FIFO2.3 CPythonqueue.Queue2.3.1 Python 有哪些 Queue2.3.2 Queue 的线程安全与同步语义2.3.3 SPSC单 Producer、单 Consumer2.3.4 MPSC多 Producer、单 Consumer2.3.5 MPMC多 Producer、多 Consumer2.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 ModelActor 拥有自身状态通过异步消息与其他 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 是更加完整的并发计算模型。模型 / 方案中间机制数据交换主要关注点Consumer 未准备好时DijkstraShared Buffer直接访问共享缓冲区数据同步与进程协作有缓冲区未满时可以写入满时等待双方可以异步交接RC 4000系统消息队列系统消息接口操作系统通信机制有缓冲区未满时可以写入满时等待双方可以异步交接Actor ModelActor向 Actor 发送消息Actor 的身份、状态与行为有缓冲区未满时可以写入满时等待双方可以异步交接CSPRendezvousSend / Receive 配对通信事件与执行同步无缓冲区必须等待接收方就绪双方同步交接1.6 回到 Message PassingMessage Passing 的核心是让执行单元通过显式的消息发送与接收交换数据、协调执行。。 通信可以有缓冲也可以无缓冲有缓冲时消息可以先暂存发送方与接收方异步交接无缓冲时消息不能暂存必须等待收发双方就绪后同步交接。接下来我们分别从 JavaBlockingQueue、GoChannel和 CPythonqueue.Queue的语言规则出发回答以下三个问题Atomicity怎样保证单次消息交接的正确性Visibility发送消息前准备的数据接收方能否正确观察Ordering发送与接收之间能够建立怎样的同步顺序2. 三种语言的消息交接规则分别保证什么2.1 JavaBlockingQueue2.1.1 Java 有哪些 QueueJava 的Queue是一种存放待处理元素的集合接口。根据是否支持并发访问、是否需要等待我们可以把常见实现分成三类类型代表实现是否支持并发访问是否提供阻塞等待操作普通 QueueArrayDeque、PriorityQueue否否并发非阻塞 QueueConcurrentLinkedQueue是否队列为空时poll()立即返回nullBlockingQueueArrayBlockingQueue等是是例如take()等待元素put()在容量受限且已满时等待空间其中BlockingQueue有几种代表实现实现存储结构容量取出规则阻塞等待条件ArrayBlockingQueue数组固定创建时指定FIFOput()在满时等待take()在空时等待LinkedBlockingQueue链表创建时可指定默认Integer.MAX_VALUEFIFOput()在满时等待take()在空时等待SynchronousQueue不存储元素0收发双方直接配对put()等待接收方take()等待发送方PriorityBlockingQueue堆无界按优先级同优先级不保证 FIFOput()不等待空间take()在空时等待DelayQueue优先级堆无界按到期时间先到期先取出put()不等待空间take()在空或尚无到期元素时等待LinkedTransferQueue链表无界FIFOput()不等待空间take()在空时等待transfer()等待元素被接收这里选用最典型的实现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。对于 VisibilityProducer 在发送消息前完成的写入对取得该消息的 Consumer 可见。对于 Ordering发送前的操作、消息交接以及接收后的操作通过 happens-before 的传递性建立跨线程内存顺序。继续沿用第零篇的count从 SPSC 扩展到 MPSC、MPMCclassIncrement{intdelta;}BlockingQueueIncrementqueuenewArrayBlockingQueue(2);2.1.3 SPSC单 Producer、单 Consumer// Producer 线程IncrementmessagenewIncrement();message.delta1;queue.put(message);// Consumer 线程intcount0;Incrementreceivedqueue.take();countreceived.delta;// count 1对应的执行关系Producer Consumer message.delta 1 │ │ program order ▼ queue.put(message) ─── 消息交接 ────────► queue.take() 返回 message │ │ program order ▼ read received.delta 1 count 1put()成功放入一条消息take()成功取出并移除它。这符合 Atomicity 的单次操作保证但此时还没有多个 Producer 或多个 Consumer 之间的竞争。delta是普通字段没有使用volatile或 Atomic。假设消息入队后没有其他线程继续修改它Producer 的字段写入 happens-before Consumer 的读取所以 Consumer 能看到1也符合 Visibility 和 Ordering 的保证。2.1.4 MPSC多 Producer、单 Consumer// Producer A、B 各自在自己的线程执行一次IncrementmessagenewIncrement();message.delta1;queue.put(message);// 唯一的 Consumer 线程intcount0;for(inti0;i2;i){countqueue.take().delta;}// 两条消息均成功提交并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer put(A: 1) put(B: 1) take() - 1 count: 0 - 1 take() - 1 count: 1 - 2Atomicity两个 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 对象IncrementmessagenewIncrement();message.delta1;queue.put(message);// Consumer A、B各执行一次Incrementreceivedqueue.take();intdeltareceived.delta;例如两条消息分别由两个 Consumer 取走Producer A Producer B Consumer A Consumer B put(A: 1) put(B: 1) take() - A take() - BAtomicity同一条消息只会被一次成功的take()移除不会被两个 Consumer 重复取走。图中 A、B 只是区分两次发送的消息对象并不表示固定分配给哪个 Consumer。Visibility / Ordering无论哪个 Consumer 取得消息都能依赖该消息对应 Producer 的入队前写入与接收后读取之间的 happens-before 关系。但消息出队的先后并不保证两个 Consumer 业务处理完成的先后。这里还存在一个边界如果两个 Consumer 取得消息后仍直接修改同一个普通count// 反例两个 Consumer 共享同一个 Counter 对象classCounter{intcount;}CountersharedCounternewCounter();// 先创建交给两个 Consumer 共用// Consumer A、B 各在自己的线程中执行一次intdeltaqueue.take().delta;sharedCounter.countdelta;两条消息虽然各自被安全取出但count仍可能因并发更新而只得到1。BlockingQueue 保证的是消息交接的原子性而不是任意后续业务操作的原子性。2.1.6 阻塞等待与 FIFO以上三个场景还共享ArrayBlockingQueue的两个具体行为阻塞等待队列为空时take()等待容量为2的队列满时put()等待可用空间。FIFO元素按照实际入队的顺序出队。但多个 Producer 的竞争入队顺序不预先确定也不能据此推断多个 Consumer 的业务完成顺序。2.2 GoChannel与 Java 不同Go 没有提供一组与BlockingQueue对应的标准并发队列实现而是直接在语言层提供了Channel用于 Goroutine 之间的数据通信。2.2.1 Go 有哪些 Channel根据 Go Language SpecificationChannel 可以按照缓冲容量分为两种类型创建方式主要特点Unbuffered Channelmake(chan int)不存储消息发送与接收需要配对Buffered Channelmake(chan int, 2)具有固定容量的缓冲区允许发送者先提交消息Channel 还可以限制通信方向类型声明允许的操作双向 Channelchan T发送和接收Send-only Channelchan- T只能发送Receive-only Channel-chan T只能接收方向限制与是否有缓冲是两个独立的维度。例如一个 Buffered Channel 也可以通过chan- T类型交给只负责发送消息的函数。这里选用典型的Buffered Channel来讲解。初始化一个容量为2的 Buffered ChanneltypeIncrementstruct{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。对于 VisibilityProducer 在发送消息前完成的写入对完成对应接收的 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 countreceived.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;i2;i{received:-jobs countreceived.delta}// 两条消息均成功发送并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer send(A: 1) send(B: 1) receive - 1 count: 0 - 1 receive - 1 count: 1 - 2Atomicity两个 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 - BAtomicity同一次发送的消息只由一个对应的 Receive 取得不会被两个 Consumer 重复接收。图中的 A、B 仅用来区分消息不代表固定的消费者分配规则。Visibility / Ordering每个 Consumer 都能依赖所接收消息的 Send / Receive 同步关系但接收消息的顺序并不保证处理完成的顺序。如果两个 Consumer 接收后仍修改同一个共享count// 反例两个 Consumer 共享同一个普通计数器received:-jobs sharedCounter.countreceived.delta即使两个 Receive 都正确取得消息sharedCounter.count仍然存在 Data Race可能发生丢失更新。Channel 保障消息交接不会自动保护接收后的共享状态。2.2.6 Buffered / Unbuffered 与 FIFO前面的例子使用容量为2的 Buffered Channel等待缓冲区空时 Receive 等待缓冲区满时 Send 等待。FIFO同一个 Channel 按消息实际发送的顺序交接但并发 Producer 的发送先后并不预先确定也不保证多个 Consumer 的业务完成顺序。Go 还支持 Unbuffered Channeljobs:make(chanIncrement)它不存储消息Send 和 Receive 必须配对Send 才能完成。Go Memory Model 还规定Unbuffered Channel 的对应 Receive synchronized-before Send 完成。这项额外保证不适用于一般的 Buffered Channel 发送完成时刻。Channel 如何让等待中的 Goroutine 挂起并在条件满足时恢复执行是下一篇要回答的问题。2.3 CPythonqueue.Queue与 Go 不同Python 没有在语言语法层提供 Channel而是通过标准库提供不同执行模型下的队列机制。2.3.1 Python 有哪些 QueuePython 为不同的执行单元提供了不同的队列模块代表类型通信双方等待方式queueQueue、SimpleQueue等同一进程内的线程阻塞当前线程asyncioasyncio.Queue同一事件循环内的异步 Task挂起当前 Task让其他 Task 继续执行multiprocessingmultiprocessing.Queue不同进程阻塞调用方其中queue提供线程安全队列asyncio.Queue面向异步 Task不提供线程安全保证multiprocessing.Queue面向多进程通信传递的对象需要经过序列化。这里讨论 CPython 的线程因此重点介绍标准库queue模块。根据 Python queue 官方文档它提供以下几种代表实现实现容量取出规则默认阻塞等待条件任务完成跟踪queue.Queue可配置默认无界FIFO先进先出put()在满时等待get()在空时等待支持queue.LifoQueue可配置默认无界LIFO后进先出put()在满时等待get()在空时等待支持queue.PriorityQueue可配置默认无界按优先级值小的先取出put()在满时等待get()在空时等待支持queue.SimpleQueue无界FIFO先进先出put()不等待空间get()在空时等待不支持任务完成跟踪指通过task_done()标记任务处理完成通过join()等待所有入队任务处理完成。注意Python 的 maxsize 0 表示缓冲区容量不设上限Go 的 Channel 容量为 0 则表示没有缓冲区。这里选用典型的实现queue.Queue来讲解。其中两个最重要的操作是操作含义默认等待条件put(item)放入元素队列满时等待可用空间get()取出并移除元素队列空时等待新元素两者默认阻塞等待也可以通过blockFalse立即尝试或通过timeout限制等待时间未能放入或取出元素时分别抛出queue.Full或queue.Empty。2.3.2 Queue 的线程安全与同步语义Python queue 官方文档 规定“TheQueueclass in this module implements all the required locking semantics.”这条契约对应Atomicityqueue.Queue支持多个线程并发执行put()和get()使用内部锁保证队列操作的线程安全同一条消息不会被两个 Consumer 重复取出。对于Visibility和OrderingProducer 在put()之前准备消息Consumer 通过get()取得该消息再读取其字段。在 CPython 中这一交接依靠queue.Queue内部的锁与条件变量完成线程同步使 Consumer 能读取交接前准备好的数据。。继续沿用第零篇的count从 SPSC 扩展到 MPSC、MPMCfromdataclassesimportdataclassfromqueueimportQueuedataclassclassIncrement:delta:intjobsQueue(maxsize2)2.3.3 SPSC单 Producer、单 ConsumerProducer 发送一条1消息唯一的 Consumer 接收后更新count# Producer ThreadmessageIncrement(delta1)jobs.put(message)# Consumer Threadcount0receivedjobs.get()countreceived.delta# count 1对应的执行关系Producer Consumer message.delta 1 │ ▼ jobs.put(message) ─── Queue 内部同步 ───► jobs.get() 返回 message │ ▼ read received.delta 1 count 1put()成功放入一条消息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(delta1))# 唯一的 Consumer Threadcount0for_inrange(2):receivedjobs.get()countreceived.delta# 两条消息均成功提交并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer put(A: 1) put(B: 1) get() - 1 count: 0 - 1 get() - 1 count: 1 - 2Atomicity两个 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(delta1))# Consumer A、B 各自在自己的 Thread 执行一次receivedjobs.get()deltareceived.delta例如两条消息分别由两个 Consumer 取得Producer A Producer B Consumer A Consumer B put(A: 1) put(B: 1) get() - A get() - BAtomicity成功的get()会从队列移除一个元素同一条消息不会被两个 Consumer 同时取出。图中的 A、B 仅用于区分消息不表示预先指定哪个 Consumer 接收。Visibility / Ordering取得消息的 Consumer 可以依赖 Queue 的线程同步读取入队前准备的数据但出队顺序不等于两个 Consumer 的业务完成顺序。如果两个 Consumer 之后仍然修改同一个共享计数器# 反例两个 Consumer 共用 countercounter{value:0}# 在创建两个 Consumer 之前共享# Consumer A、B 各自执行一次receivedjobs.get()counter[value]received.delta队列安全取出两条消息并不能保证两次counter[value] 1不发生丢失更新。不能依赖 GIL 把这个复合操作当作 Python 的原子性保证Free-threaded CPython 更不能这样假设。2.3.6 阻塞等待、FIFO 与任务完成前面的三个场景还共享queue.Queue的具体行为阻塞等待容量为2的队列满时put()默认等待空队列上的get()默认等待。maxsize0表示不限制容量而不是 Go 式无缓冲通信。FIFOqueue.Queue按实际入队顺序取出元素并发 Producer 谁先入队无法预先确定也不保证多个 Consumer 按出队顺序完成处理。任务完成get()只表示取出消息task_done()用于报告任务处理完成join()等待未完成任务计数归零。取出消息和处理完成不是一回事。Queue 底层怎样通过 Lock 与 Condition 协调等待线程是下一篇讨论的问题。3. 下一篇消息交接是怎么实现的这一篇停在语言层JavaBlockingQueue和 GoChannel通过明确的通信规则保证消息交接及其内存顺序CPythonqueue.Queue则依靠标准库的线程同步机制完成安全交接。下一篇会把这些规则转换成实现层需要解决的问题分别沿着 JavaArrayBlockingQueue、GoChannel和 CPythonqueue.Queue从同步原语、Runtime 一直看到 OS 与 CPU。本文首发于 ThinkerQAQ 的个人博客由作者本人同步发布。原文可能持续修订最新版本请以个人博客为准。 
长河君 · 2026/10/15 16:33:19 · 阅读 1.2k 并发编程（九）：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 Passing2. 三种语言的消息交接规则分别保证什么2.1 JavaBlockingQueue2.1.1 Java 有哪些 Queue2.1.2 BlockingQueue 的两条并发规则2.1.3 SPSC单 Producer、单 Consumer2.1.4 MPSC多 Producer、单 Consumer2.1.5 MPMC多 Producer、多 Consumer2.1.6 阻塞等待与 FIFO2.2 GoChannel2.2.1 Go 有哪些 Channel2.2.2 Channel 的通信与同步规则2.2.3 SPSC单 Producer、单 Consumer2.2.4 MPSC多 Producer、单 Consumer2.2.5 MPMC多 Producer、多 Consumer2.2.6 Buffered / Unbuffered 与 FIFO2.3 CPythonqueue.Queue2.3.1 Python 有哪些 Queue2.3.2 Queue 的线程安全与同步语义2.3.3 SPSC单 Producer、单 Consumer2.3.4 MPSC多 Producer、单 Consumer2.3.5 MPMC多 Producer、多 Consumer2.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 ModelActor 拥有自身状态通过异步消息与其他 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 是更加完整的并发计算模型。模型 / 方案中间机制数据交换主要关注点Consumer 未准备好时DijkstraShared Buffer直接访问共享缓冲区数据同步与进程协作有缓冲区未满时可以写入满时等待双方可以异步交接RC 4000系统消息队列系统消息接口操作系统通信机制有缓冲区未满时可以写入满时等待双方可以异步交接Actor ModelActor向 Actor 发送消息Actor 的身份、状态与行为有缓冲区未满时可以写入满时等待双方可以异步交接CSPRendezvousSend / Receive 配对通信事件与执行同步无缓冲区必须等待接收方就绪双方同步交接1.6 回到 Message PassingMessage Passing 的核心是让执行单元通过显式的消息发送与接收交换数据、协调执行。。 通信可以有缓冲也可以无缓冲有缓冲时消息可以先暂存发送方与接收方异步交接无缓冲时消息不能暂存必须等待收发双方就绪后同步交接。接下来我们分别从 JavaBlockingQueue、GoChannel和 CPythonqueue.Queue的语言规则出发回答以下三个问题Atomicity怎样保证单次消息交接的正确性Visibility发送消息前准备的数据接收方能否正确观察Ordering发送与接收之间能够建立怎样的同步顺序2. 三种语言的消息交接规则分别保证什么2.1 JavaBlockingQueue2.1.1 Java 有哪些 QueueJava 的Queue是一种存放待处理元素的集合接口。根据是否支持并发访问、是否需要等待我们可以把常见实现分成三类类型代表实现是否支持并发访问是否提供阻塞等待操作普通 QueueArrayDeque、PriorityQueue否否并发非阻塞 QueueConcurrentLinkedQueue是否队列为空时poll()立即返回nullBlockingQueueArrayBlockingQueue等是是例如take()等待元素put()在容量受限且已满时等待空间其中BlockingQueue有几种代表实现实现存储结构容量取出规则阻塞等待条件ArrayBlockingQueue数组固定创建时指定FIFOput()在满时等待take()在空时等待LinkedBlockingQueue链表创建时可指定默认Integer.MAX_VALUEFIFOput()在满时等待take()在空时等待SynchronousQueue不存储元素0收发双方直接配对put()等待接收方take()等待发送方PriorityBlockingQueue堆无界按优先级同优先级不保证 FIFOput()不等待空间take()在空时等待DelayQueue优先级堆无界按到期时间先到期先取出put()不等待空间take()在空或尚无到期元素时等待LinkedTransferQueue链表无界FIFOput()不等待空间take()在空时等待transfer()等待元素被接收这里选用最典型的实现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。对于 VisibilityProducer 在发送消息前完成的写入对取得该消息的 Consumer 可见。对于 Ordering发送前的操作、消息交接以及接收后的操作通过 happens-before 的传递性建立跨线程内存顺序。继续沿用第零篇的count从 SPSC 扩展到 MPSC、MPMCclassIncrement{intdelta;}BlockingQueueIncrementqueuenewArrayBlockingQueue(2);2.1.3 SPSC单 Producer、单 Consumer// Producer 线程IncrementmessagenewIncrement();message.delta1;queue.put(message);// Consumer 线程intcount0;Incrementreceivedqueue.take();countreceived.delta;// count 1对应的执行关系Producer Consumer message.delta 1 │ │ program order ▼ queue.put(message) ─── 消息交接 ────────► queue.take() 返回 message │ │ program order ▼ read received.delta 1 count 1put()成功放入一条消息take()成功取出并移除它。这符合 Atomicity 的单次操作保证但此时还没有多个 Producer 或多个 Consumer 之间的竞争。delta是普通字段没有使用volatile或 Atomic。假设消息入队后没有其他线程继续修改它Producer 的字段写入 happens-before Consumer 的读取所以 Consumer 能看到1也符合 Visibility 和 Ordering 的保证。2.1.4 MPSC多 Producer、单 Consumer// Producer A、B 各自在自己的线程执行一次IncrementmessagenewIncrement();message.delta1;queue.put(message);// 唯一的 Consumer 线程intcount0;for(inti0;i2;i){countqueue.take().delta;}// 两条消息均成功提交并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer put(A: 1) put(B: 1) take() - 1 count: 0 - 1 take() - 1 count: 1 - 2Atomicity两个 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 对象IncrementmessagenewIncrement();message.delta1;queue.put(message);// Consumer A、B各执行一次Incrementreceivedqueue.take();intdeltareceived.delta;例如两条消息分别由两个 Consumer 取走Producer A Producer B Consumer A Consumer B put(A: 1) put(B: 1) take() - A take() - BAtomicity同一条消息只会被一次成功的take()移除不会被两个 Consumer 重复取走。图中 A、B 只是区分两次发送的消息对象并不表示固定分配给哪个 Consumer。Visibility / Ordering无论哪个 Consumer 取得消息都能依赖该消息对应 Producer 的入队前写入与接收后读取之间的 happens-before 关系。但消息出队的先后并不保证两个 Consumer 业务处理完成的先后。这里还存在一个边界如果两个 Consumer 取得消息后仍直接修改同一个普通count// 反例两个 Consumer 共享同一个 Counter 对象classCounter{intcount;}CountersharedCounternewCounter();// 先创建交给两个 Consumer 共用// Consumer A、B 各在自己的线程中执行一次intdeltaqueue.take().delta;sharedCounter.countdelta;两条消息虽然各自被安全取出但count仍可能因并发更新而只得到1。BlockingQueue 保证的是消息交接的原子性而不是任意后续业务操作的原子性。2.1.6 阻塞等待与 FIFO以上三个场景还共享ArrayBlockingQueue的两个具体行为阻塞等待队列为空时take()等待容量为2的队列满时put()等待可用空间。FIFO元素按照实际入队的顺序出队。但多个 Producer 的竞争入队顺序不预先确定也不能据此推断多个 Consumer 的业务完成顺序。2.2 GoChannel与 Java 不同Go 没有提供一组与BlockingQueue对应的标准并发队列实现而是直接在语言层提供了Channel用于 Goroutine 之间的数据通信。2.2.1 Go 有哪些 Channel根据 Go Language SpecificationChannel 可以按照缓冲容量分为两种类型创建方式主要特点Unbuffered Channelmake(chan int)不存储消息发送与接收需要配对Buffered Channelmake(chan int, 2)具有固定容量的缓冲区允许发送者先提交消息Channel 还可以限制通信方向类型声明允许的操作双向 Channelchan T发送和接收Send-only Channelchan- T只能发送Receive-only Channel-chan T只能接收方向限制与是否有缓冲是两个独立的维度。例如一个 Buffered Channel 也可以通过chan- T类型交给只负责发送消息的函数。这里选用典型的Buffered Channel来讲解。初始化一个容量为2的 Buffered ChanneltypeIncrementstruct{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。对于 VisibilityProducer 在发送消息前完成的写入对完成对应接收的 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 countreceived.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;i2;i{received:-jobs countreceived.delta}// 两条消息均成功发送并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer send(A: 1) send(B: 1) receive - 1 count: 0 - 1 receive - 1 count: 1 - 2Atomicity两个 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 - BAtomicity同一次发送的消息只由一个对应的 Receive 取得不会被两个 Consumer 重复接收。图中的 A、B 仅用来区分消息不代表固定的消费者分配规则。Visibility / Ordering每个 Consumer 都能依赖所接收消息的 Send / Receive 同步关系但接收消息的顺序并不保证处理完成的顺序。如果两个 Consumer 接收后仍修改同一个共享count// 反例两个 Consumer 共享同一个普通计数器received:-jobs sharedCounter.countreceived.delta即使两个 Receive 都正确取得消息sharedCounter.count仍然存在 Data Race可能发生丢失更新。Channel 保障消息交接不会自动保护接收后的共享状态。2.2.6 Buffered / Unbuffered 与 FIFO前面的例子使用容量为2的 Buffered Channel等待缓冲区空时 Receive 等待缓冲区满时 Send 等待。FIFO同一个 Channel 按消息实际发送的顺序交接但并发 Producer 的发送先后并不预先确定也不保证多个 Consumer 的业务完成顺序。Go 还支持 Unbuffered Channeljobs:make(chanIncrement)它不存储消息Send 和 Receive 必须配对Send 才能完成。Go Memory Model 还规定Unbuffered Channel 的对应 Receive synchronized-before Send 完成。这项额外保证不适用于一般的 Buffered Channel 发送完成时刻。Channel 如何让等待中的 Goroutine 挂起并在条件满足时恢复执行是下一篇要回答的问题。2.3 CPythonqueue.Queue与 Go 不同Python 没有在语言语法层提供 Channel而是通过标准库提供不同执行模型下的队列机制。2.3.1 Python 有哪些 QueuePython 为不同的执行单元提供了不同的队列模块代表类型通信双方等待方式queueQueue、SimpleQueue等同一进程内的线程阻塞当前线程asyncioasyncio.Queue同一事件循环内的异步 Task挂起当前 Task让其他 Task 继续执行multiprocessingmultiprocessing.Queue不同进程阻塞调用方其中queue提供线程安全队列asyncio.Queue面向异步 Task不提供线程安全保证multiprocessing.Queue面向多进程通信传递的对象需要经过序列化。这里讨论 CPython 的线程因此重点介绍标准库queue模块。根据 Python queue 官方文档它提供以下几种代表实现实现容量取出规则默认阻塞等待条件任务完成跟踪queue.Queue可配置默认无界FIFO先进先出put()在满时等待get()在空时等待支持queue.LifoQueue可配置默认无界LIFO后进先出put()在满时等待get()在空时等待支持queue.PriorityQueue可配置默认无界按优先级值小的先取出put()在满时等待get()在空时等待支持queue.SimpleQueue无界FIFO先进先出put()不等待空间get()在空时等待不支持任务完成跟踪指通过task_done()标记任务处理完成通过join()等待所有入队任务处理完成。注意Python 的 maxsize 0 表示缓冲区容量不设上限Go 的 Channel 容量为 0 则表示没有缓冲区。这里选用典型的实现queue.Queue来讲解。其中两个最重要的操作是操作含义默认等待条件put(item)放入元素队列满时等待可用空间get()取出并移除元素队列空时等待新元素两者默认阻塞等待也可以通过blockFalse立即尝试或通过timeout限制等待时间未能放入或取出元素时分别抛出queue.Full或queue.Empty。2.3.2 Queue 的线程安全与同步语义Python queue 官方文档 规定“TheQueueclass in this module implements all the required locking semantics.”这条契约对应Atomicityqueue.Queue支持多个线程并发执行put()和get()使用内部锁保证队列操作的线程安全同一条消息不会被两个 Consumer 重复取出。对于Visibility和OrderingProducer 在put()之前准备消息Consumer 通过get()取得该消息再读取其字段。在 CPython 中这一交接依靠queue.Queue内部的锁与条件变量完成线程同步使 Consumer 能读取交接前准备好的数据。。继续沿用第零篇的count从 SPSC 扩展到 MPSC、MPMCfromdataclassesimportdataclassfromqueueimportQueuedataclassclassIncrement:delta:intjobsQueue(maxsize2)2.3.3 SPSC单 Producer、单 ConsumerProducer 发送一条1消息唯一的 Consumer 接收后更新count# Producer ThreadmessageIncrement(delta1)jobs.put(message)# Consumer Threadcount0receivedjobs.get()countreceived.delta# count 1对应的执行关系Producer Consumer message.delta 1 │ ▼ jobs.put(message) ─── Queue 内部同步 ───► jobs.get() 返回 message │ ▼ read received.delta 1 count 1put()成功放入一条消息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(delta1))# 唯一的 Consumer Threadcount0for_inrange(2):receivedjobs.get()countreceived.delta# 两条消息均成功提交并被处理后count 2一种可能的执行顺序Producer A Producer B Consumer put(A: 1) put(B: 1) get() - 1 count: 0 - 1 get() - 1 count: 1 - 2Atomicity两个 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(delta1))# Consumer A、B 各自在自己的 Thread 执行一次receivedjobs.get()deltareceived.delta例如两条消息分别由两个 Consumer 取得Producer A Producer B Consumer A Consumer B put(A: 1) put(B: 1) get() - A get() - BAtomicity成功的get()会从队列移除一个元素同一条消息不会被两个 Consumer 同时取出。图中的 A、B 仅用于区分消息不表示预先指定哪个 Consumer 接收。Visibility / Ordering取得消息的 Consumer 可以依赖 Queue 的线程同步读取入队前准备的数据但出队顺序不等于两个 Consumer 的业务完成顺序。如果两个 Consumer 之后仍然修改同一个共享计数器# 反例两个 Consumer 共用 countercounter{value:0}# 在创建两个 Consumer 之前共享# Consumer A、B 各自执行一次receivedjobs.get()counter[value]received.delta队列安全取出两条消息并不能保证两次counter[value] 1不发生丢失更新。不能依赖 GIL 把这个复合操作当作 Python 的原子性保证Free-threaded CPython 更不能这样假设。2.3.6 阻塞等待、FIFO 与任务完成前面的三个场景还共享queue.Queue的具体行为阻塞等待容量为2的队列满时put()默认等待空队列上的get()默认等待。maxsize0表示不限制容量而不是 Go 式无缓冲通信。FIFOqueue.Queue按实际入队顺序取出元素并发 Producer 谁先入队无法预先确定也不保证多个 Consumer 按出队顺序完成处理。任务完成get()只表示取出消息task_done()用于报告任务处理完成join()等待未完成任务计数归零。取出消息和处理完成不是一回事。Queue 底层怎样通过 Lock 与 Condition 协调等待线程是下一篇讨论的问题。3. 下一篇消息交接是怎么实现的这一篇停在语言层JavaBlockingQueue和 GoChannel通过明确的通信规则保证消息交接及其内存顺序CPythonqueue.Queue则依靠标准库的线程同步机制完成安全交接。下一篇会把这些规则转换成实现层需要解决的问题分别沿着 JavaArrayBlockingQueue、GoChannel和 CPythonqueue.Queue从同步原语、Runtime 一直看到 OS 与 CPU。本文首发于 ThinkerQAQ 的个人博客由作者本人同步发布。原文可能持续修订最新版本请以个人博客为准。