Python 实现 ZeroMQ 三种核心工作模式详解

ZeroMQ 简介与核心特性

简介

官方对 ZMQ（ZeroMQ 简称）的描述指出，它是一个轻量级的传输层库，类似于框架的 Socket 库，旨在简化 Socket 编程，使其更简洁且具备更高的性能。

作为一个消息处理队列库，ZMQ 支持在多个线程、内核以及不同主机之间进行弹性伸缩通信。

ZMQ 的明确目标是“成为标准网络协议栈的一部分，并进入 Linux 内核”。尽管目前尚未实现这一目标，但它无疑是一个极具前景的项目，是对传统 BSD 套接字的高效封装，使得编写高性能网络应用程序变得简单且有趣。

ZeroMQ 与 RabbitMQ、ActiveMQ 等消息队列服务器有本质区别。它并非一个独立的消息队列服务器，而是一组底层网络通信库，对原有 Socket API 进行了封装，从而简化了操作。

ZeroMQ 三种基本工作模式详解

三种工作模式

Request-Reply 模式：请求与应答机制

Request-Reply 模式：

“请求-应答”模式的核心在于其消息流动模型，该模式严格遵循“一问一答”的交互方式。

在发出消息后，若未收到回复，再次发送第二条消息时会抛出异常。同样地，对于 Rep 端，在未接收到消息前，不允许发送消息。

基于此机制，构成了“一问一答”的响应模式。

server:

# -*- coding=utf-8 -*-

import zmq

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

while True:
 message = socket.recv()
 print("Received: %s" % message)
 socket.send("I am OK!")

client:

# -*- coding=utf-8 -*-

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

socket.send('Are you OK?')
response = socket.recv()
print("response: %s" % response)

Publish-Subscribe 模式：发布与订阅机制

Publish-Subscribe 模式：

在“发布-订阅”模式下，“发布者”绑定一个指定地址（如“192.168.10.1：5500”），“订阅者”连接至该地址。消息流是单向的，仅允许从“发布者”流向“订阅者”。此外，“发布者”只管发送消息，不关心是否存在“订阅者”。一个“发布者”可以拥有多个订阅者，反之，一个“订阅者”也可订阅多个发布者。

虽然“发布者”在发送消息时不关心“订阅者”的存在与否，但先启动“发布者”再启动“订阅者”容易导致部分消息丢失。有人可能认为“先启动‘订阅者’，再启动‘发布者’”能解决此问题。

然而，这种做法并不能保证 100% 的可靠性。在 ZeroMQ 中，存在“慢木匠”现象：即使先启动“订阅者”，再启动“发布者”，“订阅者”仍会丢失第一批数据。这是因为“订阅者”与端点建立 TCP 连接时需几毫秒的握手时间，加之 ZeroMQ 后台 IO 以异步方式执行，若双方之间未施加同步策略，消息丢失不可避免。

关于“发布-订阅”模式在 ZeroMQ 中的其他特点：

• 公平排队：当一个“订阅者”连接到多个发布者时，它会均衡地从每个“发布者”读取消息，不会出现一个“发布者”淹没其他“发布者”的情况。
• ZMQ3.0 以上版本，过滤规则发生在“发布方”；ZMQ3.0 以下版本，过滤规则发生在“订阅方”。这实际上是指处理消息的位置。

server:

# -*- coding=utf-8 -*-

import zmq
import time

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

for i in range(10):
 print('send message...' + str(i))
 socket.send('message' + str(i))
 time.sleep(1)

client:

# -*- coding=utf-8 -*-

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE, '')
while True:
 response = socket.recv()
 print("response: %s" % response)

Parallel Pipeline 模式：管道与任务分发

Parallel Pipeline 模式：

在说明“管道模式”前，需明确 ZeroMQ 中并无绝对的服务端与客户端之分，所有数据接收与发送均以连接为单位，仅区分 ZeroMQ 定义的类型。如同套接字绑定地址时，既可使用 bind，也可使用 connect，通常我们将理解中的服务端 bind 到地址，而客户端 connect 到该地址。

“管道模式”通常用于任务分发与结果收集，由一个任务发生器产生任务，“公平”地派发给其管辖下的所有 worker，完成后由结果收集器回收任务的执行结果。

整体流程易于理解：worker 连接到任务发生器上，等待任务产生，完成后将结果发送至结果收集器。若以客户端服务端概念区分，任务发生器与结果收集器视为服务端，而 worker 视为客户端。

前文提到任务派发是“公平的”，因为内部采用 LRU 算法找到最近最久未工作的闲置 worker。但此处的公平是相对的：当任务发生器启动后，第一个连接到它的 worker 会在瞬间承受整个任务发生器产生的 tasks。

总结来说，该模式由三部分组成：push 进行数据推送，work 进行数据缓存，pull 进行数据竞争获取处理。区别于 Publish-Subscribe 模式，此处存在一个数据缓存和处理负载。

当连接断开时，数据不会丢失，重连后数据继续发送到对端。

server:

# -*- coding=utf-8 -*-

import zmq
import time

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

for i in range(10):
 socket.send('message' + str(i))
 # 没启 worker 时不会发消息
 print('send message...' + str(i))
 time.sleep(1)

work:

# -*- coding=utf-8 -*-
import zmq
context = zmq.Context()
receive = context.socket(zmq.PULL)
receive.connect('tcp://127.0.0.1:5557')

sender = context.socket(zmq.PUSH)
sender.connect('tcp://127.0.0.1:5558')

while True:
 data = receive.recv()
 print('transform...' + data)
 sender.send(data)

client:

# -*- coding=utf-8 -*-
import zmq

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

while True:
 response = socket.recv()
 print("response: %s" % response)

以上。

参考文档：

https://www.jb51.net/article/177043.htm

总结

总结

到此这篇关于详解Python 实现 ZeroMQ 的三种基本工作模式的文章就介绍到这了,更多相关python ZeroMQ工作模式内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！