ZeroMQ (0MQ) 入门 - 几种经典的模式

Michael阿明

发布于 2026-03-25 13:34:49

220

1. ZeroMQ 简介
- 什么是 ZeroMQ？
- 为什么选择 ZeroMQ？
- ZeroMQ 与传统消息队列的区别
- ZeroMQ 的主要特点
2. ZeroMQ 的基本通信模式
- 请求-响应模式 (Request-Reply)
- 发布-订阅模式 (Publish-Subscribe)
- 推拉模式 (Push-Pull)
- 独家对模式 (Exclusive Pair)
3. 安装与环境设置
- 验证安装
4. 基本通信模式详解与代码实例
- 4.1 请求-响应模式 (Request-Reply)
- 4.2 发布-订阅模式 (Publish-Subscribe)
- 4.3 推拉模式 (Push-Pull)
- 4.4 独家对模式 (Exclusive Pair)
5. 实际应用案例
- 5.1 简单的分布式系统
6. 要点回顾

1. ZeroMQ 简介

什么是 ZeroMQ？

ZeroMQ (也写作 ØMQ, 0MQ 或 ZMQ) 是一个高性能的异步消息传递库，旨在用于分布式或并发应用程序。它提供了一个消息队列，无需一个专门的消息代理服务器。

为什么选择 ZeroMQ？

轻量级: 没有中心化的代理服务器，减少了系统的复杂性
高性能: 专注于低延迟和高吞吐量
灵活多样: 支持多种消息传递模式
跨平台: 可在多种操作系统上运行
多语言支持: 有多种编程语言的绑定
易于使用: 简单的 Socket 风格 API

ZeroMQ 与传统消息队列的区别

与 RabbitMQ、Kafka 等传统消息队列系统不同，ZeroMQ:

不需要单独运行的服务器
直接嵌入到应用程序中
通信延迟更低
更容易部署和维护
通信模式更加灵活

ZeroMQ 的主要特点

套接字类型: 提供多种套接字类型对应不同通信模式
消息缓存: 自动处理消息缓冲
连接处理: 自动重连机制
多部分消息: 支持发送多部分消息
传输协议: 支持多种传输协议 (tcp, ipc, inproc, pgm, epgm)

2. ZeroMQ 的基本通信模式

ZeroMQ 提供了多种通信模式，每种模式都适用于不同的应用场景。了解这些模式是有效使用 ZeroMQ 的关键。

请求-响应模式 (Request-Reply)

这是最简单也是最常见的通信模式，类似于客户端-服务器模型。客户端发送请求，服务器处理请求并回复响应。

特点:

同步通信（客户端在收到响应前会被阻塞）
一对一通信
每个请求必须等待上一个请求完成

适用场景:

Web API 调用
远程过程调用 (RPC)
需要同步请求-响应的任何场景

发布-订阅模式 (Publish-Subscribe)

在这种模式中，发布者发送消息，不关心谁接收；多个订阅者可以订阅感兴趣的消息类型。

特点:

异步通信
一对多通信
消息可以基于主题/类别过滤
发布者不关心订阅者的存在
晚连接的订阅者会错过之前发布的消息

适用场景:

实时数据广播
日志分发
事件通知系统
实时监控

推拉模式 (Push-Pull)

这种模式创建了一个管道，数据从推送端流向拉取端。多个节点可以参与推送或拉取过程，实现负载均衡。

特点:

单向数据流
可以实现并行处理和结果收集
自动负载均衡（任务按可用工作节点分配）
适合创建处理管道

适用场景:

任务分配系统
并行计算
数据处理管道
批量任务处理

独家对模式 (Exclusive Pair)

这是一种特殊模式，用于两个套接字之间的独占通信。

特点:

双向通信
一对一关系
只能有两个端点
主要用于进程内通信或特定的互连场景

适用场景:

线程间通信
特定的对等通信需求
需要双向异步通信的简单场景

3. 安装与环境设置

sudo apt-get update
sudo apt-get install libzmq3-dev
pip install pyzmq

验证安装

创建一个简单的 Python 脚本测试安装:

import zmq
print(f"Current ZMQ version: {zmq.zmq_version()}")
print(f"Current PyZMQ version: {zmq.__version__}")

如果一切顺利，你应该能看到安装的 ZMQ 和 PyZMQ 版本号。

Current ZMQ version: 4.3.5
Current PyZMQ version: 26.2.0

4. 基本通信模式详解与代码实例

现在我们已经了解了 ZeroMQ 的基础知识和不同的通信模式，接下来让我们通过实际的代码示例来深入了解每种模式。

4.1 请求-响应模式 (Request-Reply)

请求-响应是最基本的通信模式，类似于传统的客户端-服务器模型。一个请求对应一个响应，通信是同步的。

工作原理

客户端（REQ 套接字）向服务器发送请求
服务器（REP 套接字）接收请求并处理
服务器向客户端发送响应
客户端接收响应

重要的是，REQ 和 REP 套接字都遵循严格的收发顺序：

REQ: 发送-接收-发送-接收...
REP: 接收-发送-接收-发送...

如果不遵循这种模式，会导致错误。

代码实例

下面是一个简单的请求-响应模式示例，包含服务器和客户端：

# 服务器代码
import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")

while True:
    # 等待客户端请求
    message = socket.recv_string()
    print(f"收到请求: {message}")
    
    # 模拟处理请求的时间
    time.sleep(1)
    
    # 发送响应
    socket.send_string(f"对 '{message}' 的响应")

# 客户端代码
import zmq

context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5555")

for request_num in range(5):
    request = f"请求 #{request_num}"
    print(f"发送请求: {request}")
    socket.send_string(request)
    
    # 等待响应
    response = socket.recv_string()
    print(f"收到响应: {response}")

注意事项

每个请求必须等待上一个响应
如果服务器崩溃，客户端将被阻塞，注意要先启动服务端
必须严格遵循发送-接收顺序
为避免单点故障，可以实现更高级的模式，如带有心跳的请求-响应

4.2 发布-订阅模式 (Publish-Subscribe)

发布-订阅模式是一种单向数据分发模式，发布者不需要知道谁在接收消息，订阅者只获取自己感兴趣的消息。

工作原理

发布者（PUB 套接字）发送消息，通常包含一个主题/类别前缀
订阅者（SUB 套接字）通过设置过滤器来订阅感兴趣的主题
订阅者只接收与其订阅过滤器匹配的消息

代码实例

下面是一个发布-订阅模式的示例：

# 发布者代码
import zmq
import time
import random

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5556")

# 消息类别
categories = ['天气', '交通', '新闻']

while True:
    category = random.choice(categories)
    update = f"{category} 更新 #{random.randint(1, 100)}"
    socket.send_string(update)
    print(f"发布: {update}")
    time.sleep(1)

# 订阅者代码
import zmq
import sys

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

# 设置订阅过滤器（只接收"天气"相关消息）
# 留空表示接收所有消息
socket.setsockopt_string(zmq.SUBSCRIBE, "天气")

while True:
    message = socket.recv_string()
    print(f"收到: {message}")

注意事项

晚连接的订阅者会错过之前发布的消息
如果没有订阅者，发布的消息将被丢弃
必须使用 socket.setsockopt_string(zmq.SUBSCRIBE, "") 订阅所有消息
发布-订阅模式不保证消息投递
过滤是基于前缀匹配的，不是正则表达式 socket.setsockopt_string(zmq.SUBSCRIBE, "天") 也能收到天气主题

4.3 推拉模式 (Push-Pull)

推拉模式创建了一个管道，数据从一端流向另一端，类似于流水线。多个节点可以参与推送或拉取过程，实现负载均衡。

工作原理

生产者（PUSH 套接字）推送任务/数据
工作者（PULL 套接字）拉取任务/数据
多个工作者可以从同一个生产者拉取任务，实现负载均衡
结果可以通过另一个管道收集

代码实例

以下是一个推拉模式的示例，实现了一个简单的任务分配系统：

# 任务生成器代码
import zmq
import time
import random

context = zmq.Context()
socket = context.socket(zmq.PUSH)
socket.bind("tcp://*:5557")

print("开始发送任务...")
for task_id in range(100):
    workload = random.randint(1, 10)  # 随机工作负载
    socket.send_json({"id": task_id, "workload": workload})
    print(f"发送任务 #{task_id}")
    time.sleep(0.1)

# 工作者代码
import zmq
import time

context = zmq.Context()
receiver = context.socket(zmq.PULL)
receiver.connect("tcp://localhost:5557")

sender = context.socket(zmq.PUSH)
sender.connect("tcp://localhost:5558")  # 连接到结果收集器

while True:
    task = receiver.recv_json()
    print(f"处理任务 #{task['id']}")
    
    # 模拟工作
    time.sleep(task['workload'] * 0.1)
    
    # 发送结果
    sender.send_json({"task_id": task['id'], "status": "completed"})

# 结果收集器代码
import zmq

context = zmq.Context()
socket = context.socket(zmq.PULL)
socket.bind("tcp://*:5558")

while True:
    result = socket.recv_json()
    print(f"任务 #{result['task_id']} 完成")

注意事项

推拉模式是单向的
负载自动在可用工作者之间均衡分配
可以与其他模式结合使用，构建复杂的数据流
推送端在没有接收者时会阻塞

4.4 独家对模式 (Exclusive Pair)

独家对模式是一种特殊模式，用于两个套接字之间的独占双向通信。

工作原理

两个节点之间建立双向通信通道
每个节点都可以随时发送和接收消息
这种模式主要用于进程内通信或特定的互连场景

代码实例

下面是一个使用独家对模式的简单示例：

# 节点1代码
import zmq
import time
import threading

context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://*:5563")

def receive_messages():
    while True:
        message = socket.recv_string()
        print(f"节点1收到: {message}")

# 启动接收线程
thread = threading.Thread(target=receive_messages, daemon=True)
thread.start()

# 发送消息
for i in range(5):
    socket.send_string(f"从节点1的消息 {i}")
    time.sleep(1)

# 节点2代码
import zmq
import time
import threading

context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://localhost:5563")

def receive_messages():
    while True:
        message = socket.recv_string()
        print(f"节点2收到: {message}")

# 启动接收线程
thread = threading.Thread(target=receive_messages, daemon=True)
thread.start()

# 发送消息
for i in range(5):
    socket.send_string(f"从节点2的消息 {i}")
    time.sleep(1)

注意事项

PAIR 套接字只能与一个对等方连接
它不执行自动重连
必须自己处理连接状态
主要用于进程内或线程间通信

5. 实际应用案例

让我们看一些 ZeroMQ 在实际项目中的应用案例。

5.1 简单的分布式系统

这个例子展示了如何构建一个简单的分布式计算系统，用于并行处理大数据集：

# 主节点代码 (任务分配器)
import zmq

def master():
    context = zmq.Context()
    
    # 任务分发套接字
    task_sender = context.socket(zmq.PUSH)
    task_sender.bind("tcp://*:5557")
    
    # 结果收集套接字
    results_receiver = context.socket(zmq.PULL)
    results_receiver.bind("tcp://*:5558")
    
    # 控制套接字 (PUB-SUB 模式)
    controller = context.socket(zmq.PUB)
    controller.bind("tcp://*:5559")
    
    # 分发数据块任务
    data_chunks = 100
    for i in range(data_chunks):
        task_sender.send_json({"chunk_id": i, "data": f"数据块 {i}"})
    
    # 收集和处理结果
    processed_chunks = 0
    results = []
    while processed_chunks < data_chunks:
        result = results_receiver.recv_json()
        results.append(result)
        processed_chunks += 1
        print(f"进度: {processed_chunks}/{data_chunks}")
    
    # 通知工作者完成并退出
    controller.send_string("FINISHED")
    
    # 处理最终结果
    print(f"处理完成，共 {len(results)} 个结果")
    return results

if __name__ == "__main__":
    results = master()
    print("主节点处理结果:", results)

# 工作者节点代码
import zmq
import time
import random
import uuid
import multiprocessing
import signal
import sys

def worker(worker_id):
    context = zmq.Context()
    # 连接到任务分发器
    task_receiver = context.socket(zmq.PULL)
    task_receiver.connect("tcp://localhost:5557")
    
    # 连接到结果收集器
    results_sender = context.socket(zmq.PUSH)
    results_sender.connect("tcp://localhost:5558")
    
    # 连接到控制通道
    controller = context.socket(zmq.SUB)
    controller.connect("tcp://localhost:5559")
    controller.setsockopt_string(zmq.SUBSCRIBE, "")
    
    # 设置轮询器监听多个套接字
    poller = zmq.Poller()
    poller.register(task_receiver, zmq.POLLIN)
    poller.register(controller, zmq.POLLIN)
    
    print(f"工作者 {worker_id} 已启动")
    
    # 工作循环
    while True:
        try:
            socks = dict(poller.poll(timeout=1000))  # 1秒超时
            
            # 检查控制消息
            if controller in socks and socks[controller] == zmq.POLLIN:
                message = controller.recv_string(zmq.NOBLOCK)
                if message == "FINISHED":
                    break  # 退出循环
            
            # 处理任务
            if task_receiver in socks and socks[task_receiver] == zmq.POLLIN:
                try:
                    task = task_receiver.recv_json(zmq.NOBLOCK)
                    
                    # 处理数据块
                    result = process_data_chunk(task, worker_id)
                    
                    # 发送结果
                    results_sender.send_json(result)
                except zmq.Again:
                    # 没有可用任务，继续循环
                    continue
                    
        except KeyboardInterrupt:
            break
        except Exception as e:
            print(f"工作者 {worker_id} 发生错误: {e}")
            break
    
    # 清理资源
    task_receiver.close()
    results_sender.close()
    controller.close()
    context.term()
    print(f"工作者 {worker_id} 关闭")

def process_data_chunk(task, worker_id):
    # 实际数据处理逻辑
    chunk_id = task["chunk_id"]
    data = task["data"]
    
    # 模拟处理
    time.sleep(random.random() * 0.1)
    
    return {
        "chunk_id": chunk_id,
        "result": f"已处理 {data}",
        "worker_id": f"{worker_id}-{uuid.uuid4().hex[:8]}"  # 包含进程ID的工作者ID
    }

def signal_handler(signum, frame):
    print("接收到终止信号，正在关闭所有工作进程...")
    sys.exit(0)

def main():
    # 设置信号处理
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)
    
    # 获取CPU核心数，创建对应数量的工作进程
    num_processes = multiprocessing.cpu_count()
    print(f"启动 {num_processes} 个工作进程")
    
    processes = []
    
    try:
        # 创建并启动工作进程
        for i in range(num_processes):
            p = multiprocessing.Process(target=worker, args=(f"Worker-{i}",))
            p.start()
            processes.append(p)
        
        # 等待所有进程完成
        for p in processes:
            p.join()
            
    except KeyboardInterrupt:
        print("正在终止所有工作进程...")
        for p in processes:
            if p.is_alive():
                p.terminate()
                p.join(timeout=5)
                if p.is_alive():
                    p.kill()
    
    print("所有工作进程已关闭")

if __name__ == "__main__":
    multiprocessing.set_start_method('spawn', force=True)  # 确保跨平台兼容性
    main()

6. 要点回顾

ZeroMQ 是一个高性能的消息传递库，无需中央代理服务器
支持多种通信模式：请求-响应、发布-订阅、推拉、独家对
提供多语言支持和跨平台能力
具有高级功能如多部分消息、安全加密和高可用模式
适用于构建分布式系统、服务架构和实时数据处理应用

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-06-11，如有侵权请联系 cloudcommunity@tencent.com 删除

工作

入门

通信

zeromq

本文分享自 Michael阿明微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度