8. 生产者-消费者

概念

生产者和消费者速度不会永远一样。中间放一个有界队列，生产者快了就在队列满时等待，消费者快了就在队列空时等待。两边只依赖队列，不直接调用对方。这是后面消息队列、日志采集和任务调度的最小模型。队列容量是背压：容量用尽时，生产者停下来，而不是无限堆积任务。

一个队列，一端 put，另一端 take。
Java 使用 ArrayBlockingQueue，Python 使用 queue.Queue(maxsize=...)。
用一个约定对象，例如 null 或 None，通知消费者退出。
队列本身负责等待和唤起，业务代码不必再写 wait / notify。

算法流程

下面的 UML 图说明 Java 例程和 Python 例程共同实现的流程。实线箭头表示调用或发送，虚线箭头表示返回或唤起。

有界队列交接（UML 时序图）

sequenceDiagram participant Producer as 生产者 participant Queue as 有界队列 participant Consumer as 消费者 Producer->>Queue: put 数据 alt 队列已满 Queue-->>Producer: 生产者休眠，直到有空位 end Consumer->>Queue: take alt 队列为空 Queue-->>Consumer: 消费者休眠，直到有数据 end Queue-->>Consumer: 返回一份数据，只被取走一次 Producer->>Queue: put 结束标记 Consumer->>Queue: take 到结束标记后退出

Java 例程

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

public class SimpleQueue {
    public static void main(String[] args) {
        // 容量只有 2。生产者比消费者快时，第 3 个 put 会停住，这就是背压
        final BlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(2);
        Thread producer = new Thread(new Runnable() {
            public void run() {
                try {
                    for (int i = 1; i <= 5; i++) queue.put(i); // 满了就 WAITING，直到消费者 take
                    queue.put(-1);                            // 用负数通知消费者可以退出
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        Thread consumer = new Thread(new Runnable() {
            public void run() {
                try {
                    while (true) {
                        int n = queue.take();                  // 空了就 WAITING，直到生产者 put
                        if (n < 0) break;                      // 结束标记不是业务数据
                        System.out.println("got " + n);
                    }
                } catch (InterruptedException ex) {
                    Thread.currentThread().interrupt();
                }
            }
        });
        producer.start();                                      // 两边只通过队列交接，不直接调用对方
        consumer.start();
    }
}

Python 例程

import threading
from queue import Queue

q = Queue(maxsize=2)          # 有界。满了 put 会阻塞，空了 get 会阻塞

def producer():
    for i in range(1, 6):
        q.put(i)                  # 队列里已经有 2 个还没被取走时，生产者停在这里
    q.put(None)                   # None 只作为结束标记，消费者看到就退出

def consumer():
    while True:
        item = q.get()            # 没有数据时休眠，不需要自己写 wait
        if item is None:
            break
        print("got", item)        # 一份数据只被这一个消费者取走

threading.Thread(target=producer).start()
threading.Thread(target=consumer).start()