Python zmq的三种简单模式

　　ZMQ (以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层，像框架一样的一个 socket library，他使得 Socket 编程更加简单、简洁和性能更高。

是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ 的明确目标是“成为标准网络协议栈的一部分，之后进入 Linux 内核”。

ZMQ 让编写高性能网络应用程序极为简单和有趣。

ZeroMQ并不是一个对socket的封装，不能用它去实现已有的网络协议。

它有自己的模式，不同于更底层的点对点通讯模式。

它有比tcp协议更高一级的协议。（当然ZeroMQ不一定基于TCP协议，它也可以用于进程间和进程内通讯）

zeromq 并不是类似rabbitmq消息列队，它实际上只一个消息列队组件，一个库。

zeromq的几种模式

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

客户端在请求后，服务端必须回响应

由客户端发起请求，并等待服务端响应请求。从客户端端来看，一定是一对对发收配对的；

反之，在服务端一定是收发对。服务端和客户端都可以是1：N的模型。通常把1认为是server，N认为是Client。

ZMQ可以很好的支持路由功能（实现路由功能的组件叫做Device），把1：N扩展为N：M（只需要加入若干路由节点）。

从这个模型看，更底层的端点地址是对上层隐藏的。每个请求都隐含回应地址，而应用则不关心它

服务端：

# sever.py

import zmq
import sys
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")
while True:
    try:
        print("wait for client ...")
        message = socket.recv()
        print("message from client:", message.decode('utf-8'))
        socket.send(message)
    except Exception as e:
        print('异常:',e)
        sys.exit()

客户端：

#client.py

import zmq
import sys
context = zmq.Context()
print("Connecting to server...")
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5555")
while True:

    input1 = input("请输入内容：").strip()
    if input1 == 'b':
        sys.exit()
    socket.send(input1.encode('utf-8'))

    message = socket.recv()
    print("Received reply: ", message.decode('utf-8'))

Publish-Subscribe模式（发布订阅模型）:

广播所有client，没有队列缓存，断开连接数据将永远丢失。client可以进行数据过滤。

服务端

server.py

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

while True:
    msg = input("请输入要发布的信息：").strip()
    if msg == 'b':
        sys.exit()
    socket.send(msg.encode('utf-8'))
    time.sleep(1)

客户端1

client1.py

import zmq


context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE,''.encode('utf-8'))  # 接收所有消息
while True:
    response = socket.recv().decode('utf-8');
    print("response: %s" % response)

客户端2

client2.py

import zmq
context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://localhost:5555")
socket.setsockopt(zmq.SUBSCRIBE,'123'.encode('utf-8'))  # 消息过滤  只接受123开头的信息
while True:
    response = socket.recv().decode('utf-8');
    print("response: %s" % response)

运行结果：

发布端发布以下信息（注意：b是关闭发布端的指令）：

请输入要发布的信息：hello python
请输入要发布的信息：大唐不夜城
请输入要发布的信息：123435678
请输入要发布的信息：123我爱你
请输入要发布的信息：广播模式，发布端只关心发布信息，不关心订阅端是否接收
请输入要发布的信息：b

客户端1接收的信息：

response: hello python
response: 大唐不夜城
response: 123435678
response: 123我爱你
response: 广播模式，发布端只关心发布信息，不关心订阅端是否接收

客户端2接收的信息：

response: 123435678
response: 123我爱你

Parallel Pipeline模式（管道模型）：

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

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

server.py

import zmq
import time

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

while True:
    msg = input("请输入要发布的信息：").strip()
    socket.send(msg.encode('utf-8'))
    print("已发送")
    time.sleep(1)

worker.py

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("正在转发...")
    sender.send(data)

client.py

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

while True:
    response = socket.recv().decode('utf-8')
    print("response: %s" % response)

结果：

server端：

请输入要发布的信息：hello  python
已发送
请输入要发布的信息：王者不可阻挡
已发送
请输入要发布的信息：123abc
已发送
请输入要发布的信息：

work端

正在转发...
正在转发...
正在转发...

client端：（接收第二条信息后断开，断开后重新收到的信息）

response: 123abc