python中zmq的基础三种模式

ZMQ 的三个基本模型

ZMQ 提供了三个基本的通信模型，分别是“Request-Reply “，”Publisher-Subscriber“，”Parallel Pipeline”

1. 请求应答模式（Request-Reply）（rep 和 req）
   消息双向的，有来有往，req端请求的消息，rep端必须答复给req端
2. 订阅发布模式 （pub 和 sub）
   消息单向的，有去无回的。可按照发布端可发布制定主题的消息，订阅端可订阅喜欢的主题，订阅端只会收到自己已经订阅的主题。发布端发布一条消息，可被多个订阅端同事收到。
3. push pull模式
   消息单向的，也是有去无回的。push的任何一个消息，始终只会有一个pull端收到消息.

1.ZMQ 的 Request-Reply
由 Client 发起请求，并等待 Server 回应请求。请求端发送一个简单的 hello，服务端则回应一个 world。请求端和服务端都可以是 1:N 的模型。通常把 1 认为是 Server ，N 认为是 Client 。ZMQ 可以很好的支持路由功能（实现路由功能的组件叫作 Device），把 1:N 扩展为N:M （只需要加入若干路由节点）

server为REP模式，等待消息，client为REQ模式，向server请求消息。
简单的例子
server.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'))
        input1 = raw_input("请输入内容：")
        socket.send(input1)
    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 = raw_input("请输入内容：")
    if input1 == 'b':
        sys.exit()
    socket.send(input1.encode('utf-8'))
    message = socket.recv()
    print("Received reply: ", message.decode('utf-8'))

1.服务端和客户端无论谁先启动，效果是相同的，这点不同于 Socket。
2.在服务端收到信息以前，程序是阻塞的，会一直等待客户端连接上来。
3.问一答式的。如果 Server 先 send，client 先 rev 是会报错的。
4.ZMQ 通信通信单元是消息，他除了知道 Bytes 的大小，他并不关心的消息格式。因此，你可以使用任何你觉得好用的数据格式。Xml、Protocol Buffers、Thrift、json 等等。

2.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. 与 Hello World 不同的是，Socket 的类型变成 SOCKET_PUB 和 SOCKET_SUB 类型。
2. 客户端需要$subscriber->setSockOpt (ZMQ::SOCKOPT_SUBSCRIBE, $filter);设置一个过滤值，相当于设定一个订阅频道，否则什么信息也收不到。
3. 服务器端一直不断的广播中，如果中途有 Subscriber 端退出，并不影响他继续的广播，当 Subscriber 再连接上来的时候，收到的就是后来发送的新的信息了。这对比较晚加入的，或者是中途离开的订阅者，必然会丢失掉一部分信息，这是这个模式的一个问题，所谓的 Slow joiner。稍后，会解决这个问题。
4. 但是，如果 Publisher 中途离开，所有的 Subscriber 会 hold 住，等待 Publisher 再上线的时候，会继续接受信息。

客户端 （可以开多个客户端来接收消息）
client.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开头的信息
#socket.setsockopt(zmq.SUBSCRIBE,''.encode('utf-8'))  # 接收所有消息
#两种情况  这里必须使用一种来 过滤消息 不然无法接受消息
while True:
    response = socket.recv().decode('utf-8');
    print("response: %s" % response)

3.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)