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