Python zmq的三种简单模式
ZMQ (以下 ZeroMQ 簡稱 ZMQ)是一個簡單好用的傳輸層,像框架一樣的一個 socket library,他使得 Socket 編程更加簡單、簡潔和性能更高。
是一個消息處理隊列庫,可在多個線程、內(nèi)核和主機盒之間彈性伸縮。ZMQ 的明確目標是“成為標準網(wǎng)絡協(xié)議棧的一部分,之后進入 Linux 內(nèi)核”。
ZMQ 讓編寫高性能網(wǎng)絡應用程序極為簡單和有趣。
ZeroMQ并不是一個對socket的封裝,不能用它去實現(xiàn)已有的網(wǎng)絡協(xié)議。
?
它有自己的模式,不同于更底層的點對點通訊模式。
?
它有比tcp協(xié)議更高一級的協(xié)議。(當然ZeroMQ不一定基于TCP協(xié)議,它也可以用于進程間和進程內(nèi)通訊)
?
zeromq 并不是類似rabbitmq消息列隊,它實際上只一個消息列隊組件,一個庫。
?
zeromq的幾種模式
Request-Reply模式(請求響應模型):
客戶端在請求后,服務端必須回響應
由客戶端發(fā)起請求,并等待服務端響應請求。從客戶端端來看,一定是一對對發(fā)收配對的;
反之,在服務端一定是收發(fā)對。服務端和客戶端都可以是1:N的模型。通常把1認為是server,N認為是Client。
?
ZMQ可以很好的支持路由功能(實現(xiàn)路由功能的組件叫做Device),把1:N擴展為N:M(只需要加入若干路由節(jié)點)。
從這個模型看,更底層的端點地址是對上層隱藏的。每個請求都隱含回應地址,而應用則不關心它
?
?
?
服務端:
# 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("請輸入內(nèi)容:").strip()if input1 == 'b':sys.exit()socket.send(input1.encode('utf-8'))message = socket.recv()print("Received reply: ", message.decode('utf-8'))?
?
?
Publish-Subscribe模式(發(fā)布訂閱模型):
廣播所有client,沒有隊列緩存,斷開連接數(shù)據(jù)將永遠丟失。client可以進行數(shù)據(jù)過濾。
?
服務端
server.py
?
import zmqimport time
import sys
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5555")
while True:
msg = input("請輸入要發(fā)布的信息:").strip()
if msg == 'b':
sys.exit()
socket.send(msg.encode('utf-8'))
time.sleep(1)
?
?
?
客戶端1
client1.py
import zmqcontext = 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 zmqcontext = 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)
?
?
?運行結果:
發(fā)布端發(fā)布以下信息(注意:b是關閉發(fā)布端的指令):
請輸入要發(fā)布的信息:hello python 請輸入要發(fā)布的信息:大唐不夜城 請輸入要發(fā)布的信息:123435678 請輸入要發(fā)布的信息:123我愛你 請輸入要發(fā)布的信息:廣播模式,發(fā)布端只關心發(fā)布信息,不關心訂閱端是否接收 請輸入要發(fā)布的信息:b?
?
?客戶端1接收的信息:
response: hello python response: 大唐不夜城 response: 123435678 response: 123我愛你 response: 廣播模式,發(fā)布端只關心發(fā)布信息,不關心訂閱端是否接收?
?
?客戶端2接收的信息:
response: 123435678 response: 123我愛你?
?
Parallel Pipeline模式(管道模型):?
? 由三部分組成,push進行數(shù)據(jù)推送,work進行數(shù)據(jù)緩存,pull進行數(shù)據(jù)競爭獲取處理。區(qū)別于Publish-Subscribe存在一個數(shù)據(jù)緩存和處理負載。
當連接被斷開,數(shù)據(jù)不會丟失,重連后數(shù)據(jù)繼續(xù)發(fā)送到對端。
server.py
import zmq import timecontext = zmq.Context() socket = context.socket(zmq.PUSH) socket.bind("tcp://*:5557")while True:msg = input("請輸入要發(fā)布的信息:").strip()socket.send(msg.encode('utf-8'))print("已發(fā)送")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("正在轉(zhuǎn)發(fā)...")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端:
請輸入要發(fā)布的信息:hello python 已發(fā)送 請輸入要發(fā)布的信息:王者不可阻擋 已發(fā)送 請輸入要發(fā)布的信息:123abc 已發(fā)送 請輸入要發(fā)布的信息:?
?
?work端
正在轉(zhuǎn)發(fā)... 正在轉(zhuǎn)發(fā)... 正在轉(zhuǎn)發(fā)...?
?
?client端:(接收第二條信息后斷開,斷開后重新收到的信息)
response: 123abc?
轉(zhuǎn)載于:https://www.cnblogs.com/yunwangjun-python-520/p/10777375.html
總結
以上是生活随笔為你收集整理的Python zmq的三种简单模式的全部內(nèi)容,希望文章能夠幫你解決所遇到的問題。
- 上一篇: oracle查看被锁的表和解锁
- 下一篇: Python---根据字符串导入包(im