并发编程（十）：BlockingQueue 与 Channel 的实现——从 Runtime 到 CPU – ThinkerQAQ

目次

---

0. この記事は引き続き何を答えますか?

前回の記事は、言語と標準ライブラリのルールから始まり、BlockingQueue、Channel、queue.Queue がどのように原子性、可視性、順序付けを提供するのかを説明しました。

この記事は以下に続きます: Java、Go、CPython がそれぞれランタイム、OS スレッド、CPU レイヤーでこれらの保証をどのように実装するか。

---

1. JDK は ArrayBlockingQueue をどのように実装しますか?

1.1 階層化

Java Source Code
实例：BlockingQueue.put() / take()
作用：声明消息交接与阻塞等待
        │
        ▼
JDK Class Library
实例：ArrayBlockingQueue / ReentrantLock / AQS / ConditionObject
作用：维护队列状态、互斥锁和条件等待队列
        │
        ▼
JVM Implementation（HotSpot）
实例：Unsafe.park / HotSpot Parker、Continuation
作用：提供底层线程等待与虚拟线程执行机制
        │
        ▼
OS Thread / Thread Parking
实例：Platform Thread / Parker；Virtual Thread / Carrier
作用：竞争或等待时暂停、恢复相应执行单元
        │
        ▼
x86-64 Hardware
实例：Atomic Instruction / Cache Coherence / Memory Ordering
作用：提供 Atomicity、Visibility 与 Ordering 的硬件基础

1.2 プット/テイクの完全な実装パス

JDK 内のコアプロセス:

1.3 ランタイム/言語実装層の 3 つの保証

1.3.1 原子性

ここで言う原子性とは、1 つのエンキューまたはデキューによるキュー状態の変更は、他のスレッドの操作と交差しません。。

セクション 1.2 の 2 番目の図を見てください——「ArrayBlockingQueue の内部パスの挿入/取得」。put そして take 全員が最初に呼び出されます lockInterruptibly()ロックを取得した後に実行する enqueue(e) または dequeue()。配列、インデックス、 count すべての更新はロックの保護下で完了し、他のスレッドが途中でキューを変更することはできません。

1.3.2 可視性と順序付け

ここでいう可視性とは、コンシューマは、キューに参加する前に、プロデューサによって書き込まれたデータを確認できます。; 順序付けとは、プロデューサーによるこれらの書き込みは、コンシューマーによる読み取りの前に有効になります。。

セクション 1.2 の最初の図を見てください——「ArrayBlockingQueue クロスレイヤー実装パス」。プロデューサーが先に書く message.delta = 1それから電話する jobs.put(message) そしてロックを解除します（Release）。次に、コンシューマは同じロックを取得 (Acquire) します。 take() メッセージは取り出しても読めます delta == 1。前回のロック解除は後続のロックの前に行われるため、プロデューサの書き込みはコンシューマに表示されます。notEmpty.signal() 待機しているコンシューマに通知する責任があります。

1.4 ハードウェア層の3つの保証

これらの保証は、アトミック命令、キャッシュ コヒーレンス、メモリ順序制約によって最終的に適用されます。

Atomicity
  → Atomic Instruction
  → 锁状态竞争：LOCK CMPXCHG 等原子操作

Visibility
  → Cache Coherence
  → 典型 x86-64 CPU：MESI-family（如 MESIF / MOESI）

Ordering
  → Acquire / Release 与必要的 Memory Ordering
  → x86-64 内存顺序约束 + LOCKed RMW / 编译器约束

関連する実装:ArrayBlockingQueue.java、AbstractQueuedSynchronizer.java、ReentrantLock.java。

---

2. Go ランタイムはチャネルをどのように実装しますか?

2.1 階層化

Go Source Code
实例：jobs <- message / message := <-jobs
作用：声明 Send / Receive 及同步关系
        │
        ▼
Go Implementation
实例：runtime/chan.go、chansend / chanrecv、hchan、sudog
作用：管理缓冲区、等待队列及消息配对
        │
        ▼
Go Runtime / Scheduler（G / M / P）
实例：gopark / goready
作用：暂停等待的 G，将被唤醒的 G 置为可运行状态
        │
        ▼
OS Thread / Thread Parking
实例：承载 G 的 M、内部锁竞争时的线程级等待
作用：运行 Goroutine；必要时暂停或唤醒 M
        │
        ▼
x86-64 Hardware
实例：Atomic Instruction / Cache Coherence / Memory Ordering
作用：提供 Atomicity、Visibility 与 Ordering 的硬件基础

2.2 チャネル通信の完全な実装パス

まず、バッファなしチャネルを例として、「Go チャネルのクロスレイヤー実装パス」のタイミング図を見てみましょう。

チャネルのコアコミュニケーションプロセス:

2.3 ランタイム/言語実装層の 3 つの保証

2.3.1 原子性

ここで言う原子性とは、チャネルの送受信に伴うステータスの変更とメッセージのペアリングは、他のゴルーチン操作と交差しません。。

セクション 2.2 の 2 番目の図を見てください——「チャンネルの核となるコミュニケーションプロセス」。ランタイムが最初になる hchan.lock: バッファリングがある場合に変更します buf、sendx、recvx そして qcount; バッファリングがない場合は、から開始します recvq または sendq ウェイターを見つけてメッセージの引き継ぎを完了します。通信が一時的に利用できない場合、Goroutine は通過します。 gopark 待ってロックを解除し、相手がハンドオーバーを完了するのを待ってから使用します。 goready 目覚める。

2.3.2 可視性と順序付け

ここでいう可視性とは、Consumer は受信を完了すると、送信する前にProducer によって書き込まれたデータを確認できるようになります。; 順序付けとは、これらの書き込みは、コンシューマが読み取りを受信する前に有効になります。。

セクション 2.2 の最初の図を見てください——「Go Channel クロスレイヤー実装パス」。プロデューサーが先に書く message.delta = 1その後、バッファリングされていないチャネルを通じて送信されます。sendDirect 待機中の Consumer にメッセージを渡し、Consumer は受信完了後にメッセージを読むことができます。 delta == 1。 Go メモリ モデルでは、 前に同期 対応する受信動作が完了します。バッファリングされたチャネルはメッセージを最初に保存しますが、それでもこのルールに従います。

2.4 ハードウェア層の3つの保証

チャネル ロックは、基本的なアトミック命令とメモリ順序制約によって最終的に実装されます。

Atomicity
  → Atomic Instruction
  → Runtime 内部锁的 CAS / Exchange 等原子操作

Visibility
  → Cache Coherence
  → 典型 x86-64 CPU：MESI-family（如 MESIF / MOESI）

Ordering
  → Acquire / Release 与必要的 Memory Ordering
  → 编译器约束 + x86-64 内存顺序 / LOCKed RMW

ソースコード:runtime/chan.go、runtime/proc.go、internal/runtime/atomic/atomic_amd64.s。

---

3. CPython はどのように queue.Queue を実装しますか?

3.1 階層化

Python Source Code
实例：queue.Queue.put() / get()
作用：声明消息入队、出队及等待条件
        │
        ▼
CPython Standard Library
实例：queue.Queue / threading.Condition / collections.deque
作用：维护队列状态、检查条件并通知等待者
        │
        ▼
CPython Implementation
实例：_thread.lock / PyMutex / Parking Lot
作用：实现底层锁的原子状态与线程等待
        │
        ▼
OS Thread / Thread Parking
实例：Native Thread / Semaphore 等平台等待原语
作用：锁竞争或条件等待时阻塞与恢复线程
        │
        ▼
x86-64 Hardware
实例：Atomic Instruction / Cache Coherence / Memory Ordering
作用：提供 Atomicity、Visibility 与 Ordering 的硬件基础

3.2 put/get の完全な実装パス

まず、空のキューを待つコンシューマーを例として、「CPython queue.Queue クロスレイヤー実装パス」のタイミング図を見てみましょう。

「CPython キューの put/get コア プロセス」を見てみましょう。

3.3 ランタイム/言語実装層の 3 つの保証

3.3.1 原子性

ここで言う原子性とは、一度 put または get スレッド内のキューの検査と変更は、他のスレッドのキュー操作とインターリーブしません。。

セクション 3.2 の 2 番目の図を見てください——「CPython キューの put/get コアプロセス」。put そして get まずはすべて入手してください Queue.mutexキューのステータスを確認してから実行します deque.append(item) または deque.popleft()。キューがいっぱいまたは空の場合、Condition.wait() ロックは解放されて待機します。通知を受信した後、スレッドはロックを再取得し、操作を続行する前に条件を確認します。

3.3.2 可視性と順序付け

ここでいう可視性とは、コンシューマは、キューに参加する前に、プロデューサによって書き込まれたメッセージの内容を確認できます。; 順序付けとは、これらの書き込みは、コンシューマの読み取りの前に有効になります。。

セクション 3.2 の最初の図を見てください——「CPython キュー.キューのクロスレイヤー実装パス」。プロデューサーが先に書く message.delta = 1それから電話する put(message) エンキューとリリース Queue.mutex（リリース）。 Consumer が通知を受信し、同じロックを再取得 (Acquire) した後、 get() メッセージを取り出して読んでください delta == 1。ここで、同期はロックの解放と取得によって保証されます。not_empty.notify() 待機中のスレッドへの通知を担当します。

3.4 ハードウェア層の3つの保証

さらに下を見ると、Queue.mutex 最終的には _thread.lock 成し遂げる。ボトムパス PyMutex アトミック操作はロックをめぐって競合します。ロックを取得できない場合は、パーキング ロットとシステム待機メカニズムを使用してスレッドを一時停止します。

Atomicity
  → Atomic Instruction
  → PyMutex 锁状态的 CAS / Exchange

Visibility
  → Cache Coherence
  → 典型 x86-64 CPU：MESI-family（如 MESIF / MOESI）

Ordering
  → Acquire / Release 与必要的 Memory Ordering
  → 编译器原子内存顺序 + CPU 顺序约束

ソースコード:Lib/queue.py、Lib/threading.py、Python/lock.c、Python/parking_lot.c。

---

4. 3 つの実装の共通ルーチン

容量 2 のキューを例にとると、2 つのプロデューサと 2 つのコンシューマが同時に動作する場合、次の 3 つの問題を解決する必要があります。

質問	解決しない場合	対応機構
複数のプロデューサー/コンシューマーが安全にメッセージを交換するにはどうすればよいでしょうか?	メッセージは繰り返し上書きまたは取得される可能性があります	ミューテックス / アトミック性
一時的にメッセージを引き継げない場合はどうすればよいですか?	キューが空の場合は取得できず、キューがいっぱいの場合は配置できません。待機により無駄に CPU が占有される場合もあります。	コンディション / パーク / ウェイクアップ
なぜコンシューマはプロデューサによって書き込まれたデータを読み取ることができるのでしょうか?	メッセージは正常に引き継がれましたが、読み取られた内容が正しくない可能性があります。	リリース/取得、可視化/注文

4.1 誰がチームに参加またはチームを離れることができますか? ——ミューテックス / アトミック性

最初、キューには 2 つの空きがあり、次の書き込み位置は tail = 0。 2 人のプロデューサーがこの場所を同時に読み取る場合:

Producer A                    Producer B

读取 tail = 0                 读取 tail = 0
写入 buffer[0] = A            写入 buffer[0] = B

结果：两个 Producer 写入同一槽位，其中一条消息被覆盖

問題は、書き込み位置の読み取りとキューのステータスの更新の間に、他のプロデューサーが介入することもあります。。

プロデューサー A が最初にロックを取得すると仮定して、ミューテックスをキューに追加します。

Queue 初始：[空，空]，tail = 0

Producer A                    Producer B

Lock 成功                     尝试 Lock → 等待
读取 tail = 0
写入槽位 0：A
更新 tail = 1
Unlock ──────────────────→     Lock 成功
                              重新读取 tail = 1
                              写入槽位 1：B
                              更新 tail = 0
                              Unlock

最终 Queue：[A，B]

プロデューサー B は、更新されたデータを読み取る前に、A がロックを解放するまで待つ必要があります。 tailしたがって、スロット 0 には書き込まれなくなります。誰が最初にロックを取得するかは定かではありませんが、2 つの成功したエントリが互いに上書きされることはありません。

同じ原則が、キューを同時にディスパッチする 2 つのコンシューマにも適用されます。キューのステータスの確認、メッセージの取得、ステータスの更新は同じクリティカル セクションで完了する必要があります繰り返しの削除を避けるため。 Mutex の最下層は、キュー操作のアトミック性を確保するために、アトミック操作を通じてロック ステータスを競合します。

4.2 キューが空またはいっぱいの場合はどうすればよいですか? ——コンディション/パーク/ウェイクアップ

相互排他により、ハンドオーバー中に競合が発生しないことが保証されますが、キューが空または満杯の場合は待機を解決できません。

Queue 为空 → Consumer 等待 notEmpty
Queue 已满 → Producer 等待 notFull

Consumer が最初に空のキューに遭遇した例を考えてみましょう。

Consumer                         Producer

Lock → 发现 Queue 为空
登记 notEmpty 等待
Unlock → Park
                                 Lock → 入队
                                 signal(notEmpty)
                                 Unlock
Wakeup → Lock
重新检查 Queue 状态
取出消息 → Unlock

プロデューサーがキューがいっぱいになった場合は、その逆が当てはまります。待機します。 notFullデキュー後にコンシューマーによって通知されます。

待機中にロックを解放する必要があります。解放しないと、相手はキューの状態を変更できません。他のスレッドが最初にハンドオーバーを完了する可能性があるため、ウェイクアップ後に状態を再チェックする必要があります。通知の欠落を避けるために、待機の登録とロックの解放も正しく調整する必要があります。

4.3 ハンドオーバーが完了した後、メッセージの内容を読み取ることができるのはなぜですか? ——リリース/取得

最初の 2 つの質問により、ハンドオーバーが競合しないこと、およびロックが空であるために占有されないことが保証されます。ただし、コンシューマは、プロデューサによって書き込まれたデータを送信する前に確認する必要もあります。

Producer                             Consumer

写入 message.delta = 1
         ↓
发布消息 [Release] ─── 同步交接 ───▶ 接收消息 [Acquire]
                                           ↓
                                     读取 delta == 1

Mutex を使用する Queue の場合、Consumer が同じロックを取得する前に、Producer がロックを解放 (Release) し、その後、同じロックを取得します (Acquire)。したがって、コンシューマは、メッセージを送信する前に、プロデューサが何を書いたかを見ることができます。

Go のバッファなしチャネルはメッセージ バッファを必要としませんが、送信者と受信者を直接ペアにし、送信前の書き込みが受信者に見えるようにします。

メッセージを保存するためのバッファーおよびキューのステータスと合わせて、次の 3 つのメカニズムが、境界付きブロッキング キューの中核を形成します。

消息缓冲区 + 队列状态
       +
Mutex：保证一次交接不可被打断
       +
Condition：空满时等待与唤醒
       +
Release / Acquire：保证消息可见、读写有序
       ↓
Blocking Queue

---

5. 次の記事: 2 つのコラボレーション モデルの関係

次の記事では、共有メモリとメッセージ パッシングの関係と、実際のプログラムでこれらがどのように併用されるかについて説明します。第 12 章は最終章として機能し、同時プログラミング シリーズ全体をレビューして要約します。

---

この記事は最初に ThinkerQAQ の個人ブログに公開され、同時に著者自身によって公開されました。オリジナルのテキストは継続的に改訂される可能性があります。最新版については私の個人ブログを参照してください。