如何配置ZeroMQ发布-订阅（Pub-Sub）套接字向新订阅者发送最新消息？

阿华AIGC实验室

2026-4-28

可以实现！用这两种方案搞定ZeroMQ PubSub新订阅者获取最新状态的需求

当然可以配置ZeroMQ的Pub/Sub模型来满足你的需求——让新连接的订阅者立即拿到最新的系统状态，不用等下一次状态更新。核心问题在于原生Pub套接字是无状态的，不会留存消息也不跟踪订阅者，所以我们需要借助一些额外机制来实现这个功能，下面是两种最常用的方案：

方案1：用XPUB/XSUB代理留存最新状态

这是最优雅的纯Pub/Sub生态解决方案，通过在发布者和订阅者之间加一个代理，让代理负责缓存最新的状态消息，当新订阅者连接时主动推送这条消息。

原理

代理使用XSUB套接字接收发布者的所有状态消息，同时缓存最后一条消息。
代理使用XPUB套接字向订阅者转发消息，并且XPUB可以检测到新订阅者的订阅事件（当订阅者发送订阅指令时，XPUB会收到一个通知）。
一旦检测到新订阅，代理就把缓存的最新状态推送给这个订阅者。

代码示例（Python + pyzmq）

代理代码

import zmq

context = zmq.Context()

# 绑定到发布者的XSUB套接字
xsub = context.socket(zmq.XSUB)
xsub.bind("tcp://*:5556")

# 绑定到订阅者的XPUB套接字
xpub = context.socket(zmq.XPUB)
xpub.bind("tcp://*:5557")

# 缓存最后一条系统状态
last_status = None

# 使用Poller处理双向消息
poller = zmq.Poller()
poller.register(xsub, zmq.POLLIN)
poller.register(xpub, zmq.POLLIN)

while True:
    socks = dict(poller.poll())
    
    # 处理发布者发来的消息：更新缓存并转发给所有订阅者
    if xsub in socks:
        msg = xsub.recv_multipart()
        last_status = msg
        xpub.send_multipart(msg)
    
    # 处理订阅者的订阅事件：推送缓存的最新状态
    if xpub in socks:
        event = xpub.recv()
        # event的第一个字节是1表示订阅，0表示取消订阅
        if event[0] == 1:
            if last_status is not None:
                xpub.send_multipart(last_status)

发布者代码

import zmq
import time

context = zmq.Context()
pub = context.socket(zmq.PUB)
pub.connect("tcp://localhost:5556")

# 模拟系统状态更新
current_status = b"system_status: online"
pub.send_multipart([b"system_status", current_status])
print("Published initial status")

time.sleep(5)
current_status = b"system_status: maintenance"
pub.send_multipart([b"system_status", current_status])
print("Published updated status")

# 保持运行
while True:
    time.sleep(1)

订阅者代码

import zmq

context = zmq.Context()
sub = context.socket(zmq.SUB)
sub.connect("tcp://localhost:5557")
sub.setsockopt_string(zmq.SUBSCRIBE, "system_status")

# 连接后立即收到最新状态
initial_msg = sub.recv_multipart()
print(f"Initial status received: {initial_msg[1].decode()}")

# 监听后续状态更新
while True:
    msg = sub.recv_multipart()
    print(f"Updated status received: {msg[1].decode()}")

优缺点

✅ 纯Pub/Sub架构，订阅者和发布者逻辑简单，不需要额外的请求响应代码
✅ 代理统一处理缓存，适合大规模订阅者场景
❌ 需要额外部署和维护代理服务

方案2：订阅者先请求初始状态，再订阅更新

如果不想部署代理，可以让订阅者在连接Pub套接字之前，先通过一个请求响应套接字（比如REQ/REP）向发布者获取最新状态，之后再监听Pub的实时更新。

原理

发布者同时运行一个REP套接字，负责响应订阅者的初始状态请求
订阅者先连接REP套接字获取当前状态，再连接SUB套接字监听后续更新

代码示例（Python + pyzmq）

发布者代码

import zmq
import threading
import time

context = zmq.Context()

# Pub套接字：实时推送状态更新
pub = context.socket(zmq.PUB)
pub.bind("tcp://*:5558")

# Rep套接字：响应初始状态请求
rep = context.socket(zmq.REP)
rep.bind("tcp://*:5559")

current_status = "system_status: online"

def handle_status_requests():
    """后台线程处理状态请求"""
    global current_status
    while True:
        req = rep.recv()
        if req == b"get_latest_status":
            rep.send(current_status.encode())

# 启动请求处理线程
threading.Thread(target=handle_status_requests, daemon=True).start()

# 模拟状态更新
time.sleep(3)
current_status = "system_status: maintenance"
pub.send_multipart([b"system_status", current_status.encode()])
print("Published updated status")

# 保持运行
while True:
    time.sleep(1)

订阅者代码

import zmq

context = zmq.Context()

# 第一步：请求初始状态
req = context.socket(zmq.REQ)
req.connect("tcp://localhost:5559")
req.send(b"get_latest_status")
initial_status = req.recv().decode()
print(f"Initial status received: {initial_status}")

# 第二步：订阅实时更新
sub = context.socket(zmq.SUB)
sub.connect("tcp://localhost:5558")
sub.setsockopt_string(zmq.SUBSCRIBE, "system_status")

# 监听后续更新
while True:
    msg = sub.recv_multipart()
    print(f"Updated status received: {msg[1].decode()}")