当前位置：首页 > news >正文

python--杂识--19--zmq

news 来源：原创 2025/6/13 9:01:09

ZeroMQ（ØMQ）是一个高性能的异步消息库，支持多种通信模式，适用于不同的分布式系统场景。
ZeroMQ入门—1. ZeroMQ简介中有底层原理的一些介绍
以下是 ZeroMQ 在 Python 中支持的主要模式及其核心特性：

1. Request-Reply 模式

1.1 REQ-REP（请求-应答）

特点：
- 严格同步的请求-应答模式，每个请求必须对应一个应答。
- 消息按顺序处理，适合简单 RPC 调用。
限制：
- 必须按顺序发送请求和接收应答，否则会触发 ZMQError。
适用场景：
- 客户端-服务端同步通信（如数据库查询）。

Python 示例：

# REP 端（服务端）
import zmqcontext = zmq.Context()
server = context.socket(zmq.REP)
server.bind("tcp://*:5555")while True:request = server.recv()  # 接收请求print(f"Received: {request.decode()}")server.send(b"ACK")     # 发送应答# REQ 端（客户端）
import zmqcontext = zmq.Context()
client = context.socket(zmq.REQ)
client.connect("tcp://localhost:5555")client.send(b"Hello")    # 发送请求
response = client.recv() # 接收应答（必须等待）
print(f"Response: {response.decode()}")

2. Publish-Subscribe 模式

2.1 PUB-SUB（发布-订阅）

特点：
- 发布者（PUB）广播消息，订阅者（SUB）按主题（Topic）过滤消息。
- 无状态、无确认机制，适合高吞吐量场景。
限制：
- 订阅者可能因处理速度慢而丢弃消息（受高水位标记 HWM 限制）。
适用场景：
- 实时数据流（如股票行情、日志聚合）。

Python 示例：

# PUB 端（发布者）
import zmqcontext = zmq.Context()
publisher = context.socket(zmq.PUB)
publisher.bind("tcp://*:5555")while True:publisher.send_multipart([b"NEWS", b"Hello from Publisher"])# SUB 端（订阅者）
import zmqcontext = zmq.Context()
subscriber = context.socket(zmq.SUB)
subscriber.connect("tcp://localhost:5555")
subscriber.subscribe(b"NEWS")  # 订阅指定主题while True:topic, message = subscriber.recv_multipart()print(f"Received: {message.decode()}")

3. Pipeline 模式

3.1 PUSH-PULL（推-拉）

特点：
- 负载均衡模式，PUSH 端将消息分发给多个 PULL 端。
- 消息按顺序分发，适合并行处理任务。
适用场景：
- 分布式计算、任务分发（如 MapReduce）。

Python 示例：

# PUSH 端（生产者）
import zmqcontext = zmq.Context()
pusher = context.socket(zmq.PUSH)
pusher.bind("tcp://*:5555")for i in range(10):pusher.send(f"Task {i}".encode())# PULL 端（消费者）
import zmqcontext = zmq.Context()
puller = context.socket(zmq.PULL)
puller.connect("tcp://localhost:5555")while True:message = puller.recv()print(f"Processed: {message.decode()}")

4. Exclusive Pair 模式

4.1 PAIR（独占对）

特点：
- 仅支持两个对等节点之间的直接通信，无中间代理。
- 简单高效，适用于点对点场景。
适用场景：
- 父子进程通信、嵌入式系统。

Python 示例：

# 服务器端（Server.py）
import zmqcontext = zmq.Context()
server = context.socket(zmq.PAIR)
server.bind("tcp://*:5555")while True:message = server.recv()print(f"Received: {message.decode()}")server.send(b"ACK")# 客户端（Client.py）
import zmqcontext = zmq.Context()
client = context.socket(zmq.PAIR)
client.connect("tcp://localhost:5555")client.send(b"Hello from Client")
response = client.recv()
print(f"Response: {response.decode()}")

5. Router-Dealer 模式

5.1 ROUTER-DEALER（路由-经销商）

特点：
- 支持多对多通信，ROUTER 自动路由消息到多个 DEALER。
- 消息附带地址帧，适合复杂网络拓扑。
适用场景：
- 服务网格、代理模式、负载均衡。

Python 示例：

# ROUTER 端（服务端）
import zmqcontext = zmq.Context()
server = context.socket(zmq.ROUTER)
server.bind("tcp://*:5555")while True:client_addr = server.recv()  # 接收客户端地址message = server.recv()      # 接收实际消息print(f"Received from {client_addr}: {message.decode()}")# 发送响应时需指定目标地址server.send(client_addr, zmq.SNDMORE)server.send(b"ACK")# DEALER 端（客户端）
import zmqcontext = zmq.Context()
client = context.socket(zmq.DEALER)
client.connect("tcp://localhost:5555")client.send(b"Hello")
response = client.recv()  # 直接接收单帧响应
print(f"Response: {response.decode()}")

6. 其他模式

6.1 RADIO-DISH（广播-收听）
- 类似 PUB-SUB，但支持组播（Multicast）传输，适合局域网内高效广播。
- 关键差异：
  - 传输层：
    - RADIO-DISH 使用 UDP 组播，适合局域网内高效广播（如股票行情、传感器数据）。
      PUB-SUB 默认使用 TCP，支持跨网络传输，但可能引入延迟。
  - 消息去重：
    DISH 端会自动丢弃重复消息（基于序列号），确保消息唯一性。
    SUB 端需自行处理重复消息（如通过唯一 ID 过滤）。
  - 主题过滤：
    PUB-SUB 支持多级主题（如 “A.B.C”），订阅者可按主题过滤。
    RADIO-DISH 无主题概念，所有 DISH 接收相同消息。
6.2 STREAM（流）
- 专为 TCP 协议设计，支持与传统 TCP 客户端（如浏览器）交互。

模式对比总结

模式	连接方式	消息路由	适用场景
REQ-REP	1:1	严格一对一	简单 RPC、同步通信
PUB-SUB	1:N	按主题广播	实时数据流、日志聚合
PUSH-PULL	1:N	负载均衡分发	分布式计算、任务队列
PAIR	1:1	无路由	点对点通信（如进程间）
ROUTER-DEALER	N:M	动态路由	复杂网络拓扑、服务网格
RADIO-DISH	1:N	组播广播	局域网内高效广播
STREAM	1:N	传统 TCP 交互	与浏览器等传统客户端通信