9. 生产者-消费者（订阅与推送）

概念

简单队列是一份数据只被一个消费者取走。订阅推送不同：发布者产生一条消息，每一个订阅者都要收到。做法是订阅时发给对方一个独立队列，发布时把同一条消息放进所有订阅者的队列。 这就是消息推送和发布/订阅的最小结构。订阅者慢，只堵住自己的队列；可以选择丢弃新消息，也可以让发布者等待。例程使用有界队列并在满时丢弃，避免一个慢订阅者拖住全部推送。

  • 订阅：登记一个专属队列，之后只从自己的队列 take。
  • 推送：遍历当前订阅者，把消息 offer 给每一个队列。
  • 一个订阅者取走消息，不影响其他订阅者。
  • 取消订阅时摘掉队列，避免继续推给已经离开的一方。

算法流程

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

一次发布，每人一份（UML 时序图）

sequenceDiagram participant Sub as 订阅者 participant Broker as 发布者 participant Q1 as 订阅队列 1 participant Q2 as 订阅队列 2 Sub->>Broker: subscribe Broker->>Q1: 登记专属队列 Broker-->>Sub: 返回自己的队列 Broker->>Q1: publish 时放入同一条消息 Broker->>Q2: 再放入每一个其他订阅队列 Sub->>Q1: 只从自己的队列取 Note over Q2: 别人取走自己的那份，不影响这条队列 Sub->>Broker: unsubscribe Broker->>Q1: 摘掉队列，之后不再推送

Java 例程

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

public class PubSub {
    // 订阅表会被发布线程遍历，同时被订阅线程增删，用写时复制避免遍历时加锁
    private final CopyOnWriteArrayList<BlockingQueue<String>> subs =
            new CopyOnWriteArrayList<BlockingQueue<String>>();

    public BlockingQueue<String> subscribe() {
        BlockingQueue<String> q = new ArrayBlockingQueue<String>(16); // 每个订阅者一份自己的队列
        subs.add(q);
        return q;                                 // 调用方之后只从返回的队列 take
    }

    public void unsubscribe(BlockingQueue<String> q) {
        subs.remove(q);                           // 摘掉后，后续 publish 不再推给它
    }

    public void publish(String msg) {
        for (BlockingQueue<String> q : subs) {    // 同一条消息复制进每一个订阅队列
            q.offer(msg);                         // offer 在队列满时返回 false，不阻塞其他订阅者
        }
    }
}

Python 例程

import threading
from queue import Queue, Full

class Broker:
    def __init__(self):
        self.subs = []                # 普通 list 会被多个线程增删，必须加锁
        self.lock = threading.Lock()

    def subscribe(self):
        q = Queue(maxsize=16)         # 订阅者之间不共享队列，所以都能收到同一条消息
        with self.lock:
            self.subs.append(q)
        return q                      # 订阅者自己循环 get

    def unsubscribe(self, q):
        with self.lock:
            self.subs.remove(q)       # 离开后不再收到新消息

    def publish(self, msg):
        with self.lock:
            targets = list(self.subs) # 先复制名单，推送时就不必一直占着锁
        for q in targets:
            try:
                q.put_nowait(msg)      # 这个订阅者的队列满了就丢掉他这一条
            except Full:
                pass                   # 慢的那一个不影响正在推给别人
上一个 返回列表