详解Python 实现 ZeroMQ 的三种基本工作模式
简介
引用官方说法：ZMQ（以下ZeroMQ简称ZMQ）是一个简单好用的传输层，像框架一样的一个socketlibrary，他使得Socket编程更加简单、简洁和性能更高。
是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。
ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。
它跟RabbitMQ，ActiveMQ之类有着相当本质的区别，ZeroMQ根本就不是一个消息队列服务器，更像是一组底层网络通讯库，对原有的SocketAPI加上一层封装，使我们操作更简便。
三种工作模式
Request-Reply模式：
说到“请求-应答”模式，不得不说的就是它的消息流动模型。消息流动模型指的是该模式下，必须严格遵守“一问一答”的方式。
发出消息后，若没有收到回复，再发出第二条消息时就会抛出异常。同样的，对于Rep也是，在没有接收到消息前，不允许发出消息。
基于此构成“一问一答”的响应模式。
server:
#-*-coding=utf-8-*-
importzmq
context=zmq.Context()
socket=context.socket(zmq.REP)
socket.bind("tcp://*:5555")
whileTrue:
message=socket.recv()
print("Received:%s"%message)
socket.send("IamOK!")
client:
#-*-coding=utf-8-*-
importzmq
context=zmq.Context()
socket=context.socket(zmq.REQ)
socket.connect("tcp://localhost:5555")
socket.send('AreyouOK?')
response=socket.recv()
print("response:%s"%response)
Publish-Subscribe模式：
“发布-订阅”模式下，“发布者”绑定一个指定的地址，例如“192.168.10.1：5500”，“订阅者”连接到该地址。该模式下消息流是单向的，只允许从“发布者”流向“订阅者”。且“发布者”只管发消息，不理会是否存在“订阅者”。一个“发布者”可以拥有多个订阅者，同样的，一个“订阅者”也可订阅多个发布者。
虽然我们知道“发布者”在发送消息时是不关心“订阅者”的存在于否，所以先启动“发布者”，再启动“订阅者”是很容易导致部分消息丢失的。那么可能会提出一个说法“我先启动‘订阅者'，再启动‘发布者'，就能解决这个问题了？”
对于ZeroMQ而言，这种做法也并不能保证100%的可靠性。在ZeroMQ领域中，有一个叫做“慢木匠”的术语，就是说即使我是先启动了“订阅者”，再启动“发布者”，“订阅者”总是会丢失第一批数据。因为在“订阅者”与端点建立TCP连接时，会包含几毫秒的握手时间，虽然时间短，但是是存在的。再加上ZeroMQ后台IO是以一部方式执行的，所以若不在双方之间施加同步策略，消息丢失是不可避免的。
关于“发布-订阅”模式在ZeroMQ中的一些其他特点：
- 公平排队，一个“订阅者”连接到多个发布者时，会均衡的从每个“发布者”读取消息，不会出现一个“发布者”淹没其他“发布者”的情况。
- ZMQ3.0以上的版本，过滤规则发生在“发布方”。ZMQ3.0以下的版本，过滤规则发生在“订阅方”。其实也就是处理消息的位置。
server:
#-*-coding=utf-8-*-
importzmq
importtime
context=zmq.Context()
socket=context.socket(zmq.PUB)
socket.bind("tcp://*:5555")
foriinrange(10):
print('sendmessage...'+str(i))
socket.send('message'+str(i))
time.sleep(1)
client:
#-*-coding=utf-8-*-
importzmq
context=zmq.Context()
socket=context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE,'')
whileTrue:
response=socket.recv()
print("response:%s"%response)
ParallelPipeline模式：
在说明“管道模式”前，需要明确的是在ZeroMQ中并没有绝对的服务端与客户端之分，所有的数据接收与发送都是以连接为单位的，只区分ZeroMQ定义的类型。就像套接字绑定地址时，可以使用bind，也可以使用connect，只是通常我们将理解中的服务端bind到一个地址，而理解中的客户端connec到该地址。
“管道模式”一般用于任务分发与结果收集，由一个任务发生器来产生任务，“公平”的派发到其管辖下的所有worker，完成后再由结果收集器来回收任务的执行结果。
整体流程比较好理解，worker连接到任务发生器上，等待任务的产生，完成后将结果发送至结果收集器。如果要以客户端服务端的概念来区分，这里的任务发生器与结果收集器是服务端，而worker是客户端。
前面说到了这里任务的派发是“公平的”，因为内部采用了LRU的算法来找到最近最久未工作的闲置worker。但是公平在这里是相对的，当任务发生器启动后，第一个连接到它的worker会在一瞬间承受整个任务发生器产生的tasks。
总结来说由三部分组成，push进行数据推送，work进行数据缓存，pull进行数据竞争获取处理。区别于Publish-Subscribe存在一个数据缓存和处理负载。
当连接被断开，数据不会丢失，重连后数据继续发送到对端。
server:
#-*-coding=utf-8-*-
importzmq
importtime
context=zmq.Context()
socket=context.socket(zmq.PUSH)
socket.bind("tcp://*:5557")
foriinrange(10):
socket.send('message'+str(i))
#没启worker时不会发消息
print('sendmessage...'+str(i))
time.sleep(1)
work:
#-*-coding=utf-8-*-
importzmq
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')
whileTrue:
data=receive.recv()
print('transform...'+data)
sender.send(data)
client:
#-*-coding=utf-8-*-
importzmq
context=zmq.Context()
socket=context.socket(zmq.PULL)
socket.bind("tcp://*:5558")
whileTrue:
response=socket.recv()
print("response:%s"%response)
以上。
参考文档：
https://www.nhooo.com/article/177043.htm
总结
到此这篇关于详解Python实现ZeroMQ的三种基本工作模式的文章就介绍到这了,更多相关pythonZeroMQ工作模式内容请搜索毛票票以前的文章或继续浏览下面的相关文章希望大家以后多多支持毛票票！