1、zmq介紹：創(chuàng)建和銷毀套接字：zmq.socket(), zmq.close()配置和讀取套接字：zmq.setsockopt(), zmq.getsockopt()為套接字建立連接：zmq.bind(), zmq.connect()發(fā)送和接收消息： zmq.send(), zmq.recv()注：使用zmq.bind()連接的節(jié)點(diǎn)稱之為服務(wù)端，它有著一個(gè)較為固定的網(wǎng)絡(luò)地址；使用zmq.connect()連接的節(jié)點(diǎn)稱為客戶端，其地址不固定。2、zmq消息模式：主要有三種常用模式： req/rep(請(qǐng)求答復(fù)模式)：主要用于遠(yuǎn)程調(diào)用及任務(wù)分配等。 pub/sub(訂閱模式)： 主要用于數(shù)據(jù)分發(fā)。 push/pull(管道模式)： 主要用于多任務(wù)并行。3、zmq內(nèi)置的有效綁定對(duì)：PUB and SUBREQ and REPREQ and XREPXREQ and REPXREQ and XREPXREQ and XREQXREP and XREPPUSH and PULLPAIR and PAIR4、具體消息模式舉例(1)、req/rep(請(qǐng)求/答復(fù)模式)：一對(duì)一模式，一問(wèn)一答#server服務(wù)端import zmqcontext=zmq.Context()socket=context.socket(zmq.REP) #設(shè)置socket的類型，zmq.REP答復(fù)socket.bind(“tcp://*:15000”) #綁定服務(wù)端的IP和端口while True: #循環(huán)接收客戶端發(fā)來(lái)的消息 message=socket.recv() #接收客戶端發(fā)送來(lái)的消息，注：是byte類型 print(message) socket.send_string(“copy!”) #再發(fā)回客戶端消息結(jié)果：客戶單沒(méi)請(qǐng)求一次就打印一次消息體b’request’b’request’b’request’b’request’#client客戶端import zmq, syscontext = zmq.Context()socket=context.socket(zmq.REQ) #設(shè)置socket類型，請(qǐng)求端socket.connect(“tcp://localhost:15000”) #連接服務(wù)端的IP和端口while True: data=input(“input your request:”) if data == “q”: sys.exit() socket.send_string(data) #向服務(wù)端發(fā)送消息 message=socket.recv() #接收服務(wù)端返回的消息，注：是byte類型 print(message)”””結(jié)果：沒(méi)輸入請(qǐng)求一次，就得到服務(wù)端的一次返回input your data:123b’copy!’input your data:456b’copy!'”””(2)、pub/sub(訂閱模式)：一對(duì)多模式一個(gè)發(fā)布者，多個(gè)訂閱者，訂閱者可以通過(guò)設(shè)置過(guò)濾器過(guò)濾數(shù)據(jù)。#publisher發(fā)布者import zmqcontext=zmq.Context()socket=context.socket(zmq.PUB)socket.bind(“tcp://*:15000”)while True: data = input(“input your data:”) print(data) socket.send_string(data)”””結(jié)果：循環(huán)提示輸入數(shù)據(jù)，當(dāng)輸入一次，就發(fā)送一次到訂閱者input your data:123123input your data:456456input your data:789789input your data:”””#Subscriber訂閱者import sysimport zmqcontext=zmq.Context()socket=context.socket(zmq.SUB)socket.connect(“tcp://localhost:15000″)socket.setsockopt_string(zmq.SUBSCRIBE,”)或者：socket.setsockopt_string(zmq.SUBSCRIBE,’123′) #表示只過(guò)濾出收到消息為’123’的消息或者：socket.subscribe(‘topic’) #訂閱一個(gè)主題while True: message=socket.recv() print(message)”””結(jié)果：發(fā)布者每發(fā)布一次，都能訂閱到b’123’b’456’b’789′”””(3)、push/pull(管道模式)：管道是單向的，從PUSH端單向的向PULL端單向的推送數(shù)據(jù)流。由三部分組成，push進(jìn)行數(shù)據(jù)推送，work進(jìn)行數(shù)據(jù)緩存，pull進(jìn)行數(shù)據(jù)競(jìng)爭(zhēng)獲取處理。區(qū)別于Publish-Subscribe存在一個(gè)數(shù)據(jù)緩存和處理負(fù)載。當(dāng)連接被斷開(kāi)，數(shù)據(jù)不會(huì)丟失，重連后數(shù)據(jù)繼續(xù)發(fā)送到對(duì)端。#推送端import zmqcontext=zmq.Context()socket=context.socket(zmq.PUSH) #設(shè)置socket類型PUSH推送socket.bind(“tcp://*:5557”) #綁定IP和端口while True: data=input(“input your data:”) socket.send_string(data)”””input your data:123input your data:456input your data:789″””#worker端import zmqcontext=zmq.Context()socket_receive=context.socket(zmq.PULL) #設(shè)置socket類型PULL拉取推送端的消息socket_receive.connect(“tcp://localhost:5557”) #連接推送端IP和端口socket_sender=context.socket(zmq.PUSH) #再設(shè)置一個(gè)socket類型PUSH推送socket_sender.connect(“tcp://localhost:5558”) #連接IP和端口向其推送消息while True: data=socket_receive.recv_string() #拉取接收消息 print(data) socket_sender.send_string(data) #再將消息推送出去”””123456789″””#拉取端import zmqcontext=zmq.Context()socket=context.socket(zmq.PULL) #設(shè)置socket類型PULL拉取消息socket.bind(“tcp://*:5558″) #綁定IP和端口去拉取消息while True: message=socket.recv_string() print(message)”””123456789″””