AI Library · 知识库

# 任务间传递数据：生产者与消费者通道

学习目标：掌握通过通信共享内存（Do not communicate by sharing memory; instead, share memory by communicating）的并发范式，理解 CSP 通道、阻塞队列与优雅关闭。

前置知识与范围：并发基础、线程与任务。本篇通过解耦生产者与消费者的流式数据传输，演示通道与队列的关闭信号传递。

# 问题与边界

设计一个标准的生产者-消费者流水线：

  1. 生产者任务：
    • 依次生产 3 个数据项：10, 20, 30。
    • 通过通信媒介逐个投递给消费者。
    • 生产完毕后，必须发出结束信号（如关闭通道或投递终止哨兵）。
  2. 消费者任务：
    • 在独立的并发协程/线程中监听媒介。
    • 消费接收到的所有数据项，计算总和 10 + 20 + 30 = 60。
    • 检测到通道关闭或结束信号后，安全退出循环，不得发生死锁挂起。
  3. 验证契约：主线程必须等待消费者优雅收束完毕，断言消费总和严格等于 60。

# 关键差异

语言 核心通信工具 优雅关闭机制 缓冲与背压（Backpressure）
Go CSP 原生通道 chan int 原生 close(ch)，消费者 for v := range ch 自然退出 make(chan int, bufferSize) 满缓冲阻塞
Rust std::sync::mpsc::channel 当所有发送端 Sender 被 drop 销毁时，接收端 recv() 自然返回错误退出 mpsc 为无界通道；sync_channel 提供有界背压
Java java.util.concurrent.BlockingQueue 通用惯用法：投递毒丸哨兵（Poison Pill，如 -1） ArrayBlockingQueue(capacity) 阻塞等待
Python 标准库 queue.Queue 投递终止哨兵（如 None）通知消费者跳出 Queue(maxsize=...) 原生线程安全阻塞队列

# 各语言实现

# Go

Gopackage main

import (
	"fmt"
	"sync"
)

func main() {
	ch := make(chan int, 2) // 带缓冲通道
	var wg sync.WaitGroup

	var totalSum int
	wg.Add(1)

	// 启动消费者 goroutine
	go func() {
		defer wg.Done()
		for val := range ch { // 自动循环拉取，直到通道被 close 且排空
			totalSum += val
		}
	}()

	// 生产者投递数据
	for _, x := range []int{10, 20, 30} {
		ch <- x
	}
	close(ch) // 生产完毕，关闭通道

	wg.Wait() // 等待消费者消费完毕并收束

	if totalSum != 60 {
		panic("sum mismatch")
	}

	fmt.Printf("total_sum: %d\n", totalSum)
}

Go 的 close(ch) 与 range 循环是语言级原生配合，优雅消除了哨兵对象的类型污染。

# Rust

Rustuse std::sync::mpsc;
use std::thread;

fn main() {
    let (tx, rx) = mpsc::channel();

    // 启动独立消费者线程
    let consumer_handle = thread::spawn(move || {
        let mut total_sum = 0;
        // 当 tx 离开作用域被 drop 时，迭代器自然终止
        for val in rx {
            total_sum += val;
        }
        total_sum
    });

    // 生产者发送
    for &x in &[10, 20, 30] {
        tx.send(x).unwrap();
    }
    drop(tx); // 显式销毁所有发送者，向接收端发出断开信号

    let final_sum = consumer_handle.join().unwrap();
    assert_eq!(final_sum, 60);

    println!("total_sum: {final_sum}");
}

Rust 的通道完美融入所有权语义：无需显式的 close 调用，只要所有 tx 所有权离开作用域被释放，接收端即安全知晓流的终结。

# Java

Javaimport java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class Main {
    private static final int POISON_PILL = -1; // 结束哨兵

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
        int[] totalSum = new int[]{0};

        Thread consumer = new Thread(() -> {
            try {
                while (true) {
                    int val = queue.take(); // 阻塞式拉取
                    if (val == POISON_PILL) {
                        break; // 收到毒丸，退出消费循环
                    }
                    totalSum[0] += val;
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        });
        consumer.start();

        // 生产者入队
        for (int x : new int[]{10, 20, 30}) {
            queue.put(x);
        }
        queue.put(POISON_PILL); // 投递结束标志

        consumer.join(); // 等待消费者彻底收束

        assert totalSum[0] == 60;
        System.out.println("total_sum: " + totalSum[0]);
    }
}

Java 标准库通过线程安全的 BlockingQueue 实现带背压的生产者-消费者模型。

# Python

Pythonimport queue
import threading

POISON_PILL = None

def consumer_loop(q: queue.Queue, result_box: list):
    total = 0
    while True:
        item = q.get()
        if item is POISON_PILL:
            q.task_done()
            break
        total += item
        q.task_done()
    result_box.append(total)

if __name__ == "__main__":
    q: queue.Queue = queue.Queue(maxsize=10)
    result_box = []

    t = threading.Thread(target=consumer_loop, args=(q, result_box))
    t.start()

    for x in [10, 20, 30]:
        q.put(x)
    q.put(POISON_PILL)

    t.join()

    assert result_box[0] == 60
    print(f"total_sum: {result_box[0]}")

Python 的 queue.Queue 内置锁与条件变量，天然支持线程间协调。

# 迁移误区与适用限制

# 运行与验证

语言版本基线见示例基线。 所有程序运行后控制台输出均为：

纯文本total_sum: 60

通道/队列发送、优雅关闭与汇总累加断言全部通过。

# 相关阅读