戰(zhàn))
最近在做一個分布式采集的中間件改造幾個服務(wù)之間需要頻繁傳遞任務(wù)消息最開始圖省事直接上了HTTP接口輪詢結(jié)果超時重試、連接風(fēng)暴、數(shù)據(jù)丟失輪著來。后來換成了ZMQZeroMQ這套消息通信庫才算把這塊理順。這個系列打算把我對ZMQ通信模式的理解完整梳理一遍本篇是上篇重點(diǎn)講最常用的三種基礎(chǔ)模式請求-應(yīng)答REQ/REP、發(fā)布-訂閱PUB/SUB、推拉管道PUSH/PULL以及它們各自適合什么場景、有哪些坑配合可運(yùn)行的代碼示例方便你直接照著復(fù)現(xiàn)。如果你正在做微服務(wù)通信、分布式任務(wù)調(diào)度、數(shù)據(jù)采集管道這類東西又不想引入Kafka或RabbitMQ這種重量級中間件那ZMQ會是一個非常合適的選擇。它足夠輕沒有獨(dú)立Broker進(jìn)程功能又足夠硬核理解好通信模式就等于拿到了它的使用鑰匙。1. 從裸socket到消息總線的跳躍ZMQ到底解決了什么問題1.1 傳統(tǒng)socket開發(fā)中最讓人頭疼的三件事很多人剛接觸ZMQ時容易把它當(dāng)成一個普通的網(wǎng)絡(luò)庫但實(shí)際上它解決的是如何優(yōu)雅地建立分布式消息通道這個層面的問題。用裸socket寫通信時你大概率會被三件事折磨。第一是連接管理。兩個進(jìn)程要通信你得手動處理建立連接、斷開重連、心跳?；睢＿B接斷了他不會自己恢復(fù)你需要寫一堆重試邏輯。第二是消息邊界。TCP是字節(jié)流你發(fā)了hello和world對端收到的可能是helloworld、也可能是hel和loworld你得自己設(shè)計(jì)報文頭、長度字段、粘包拆包邏輯。第三是架構(gòu)靈活性。裸socket天然是1對1的想要1對多廣播、多對多消息路由那得自己實(shí)現(xiàn)一套注冊發(fā)現(xiàn)機(jī)制。ZMQ把這三件事全部收納了。它內(nèi)置了斷線重連機(jī)制你寫的代碼里根本不需要關(guān)心連接狀態(tài)——它自己會悄悄重連。它面向消息而非字節(jié)流每條消息有天然邊界不存在粘包拆包問題。它提供多種通信模式就像給消息傳輸定義了不同規(guī)矩你照著規(guī)矩連就行。一句話理解socket給你的是字節(jié)管道ZMQ給你的是消息總線后者自帶連接管理、消息分幀和多種路由策略。1.2 ZMQ的核心抽象消息、端點(diǎn)與上下文使用ZMQ需要先理解三個基礎(chǔ)對象Context、Socket和Endpoint。Context是ZMQ的總環(huán)境所有socket都必須在一個Context中創(chuàng)建類似進(jìn)程級的資源管理器。在pyzmq中通常一行代碼搞定ctx zmq.Context()。每個socket通過socket_type指定模式比如zmq.REQ、zmq.PUB。Endpoint是socket綁定的地址ZMQ支持多種傳輸協(xié)議最常用的是tcp://跨主機(jī)、ipc://同一主機(jī)進(jìn)程間、inproc://同一進(jìn)程內(nèi)線程間。這里要特別強(qiáng)調(diào)一個設(shè)計(jì)理念ZMQ的socket不像傳統(tǒng)socket那樣區(qū)分服務(wù)端和客戶端。一個socket可以bind綁定也可以connect連接而且規(guī)則很靈活——不是只有服務(wù)端才能bind客戶端也能bind服務(wù)端也能connect。理解這一點(diǎn)對后面的模式會很關(guān)鍵。1.3 環(huán)境準(zhǔn)備安裝與第一個連接動手之前先把環(huán)境裝上。Python環(huán)境用pip安裝pyzmqpip install pyzmq安裝完成后寫一個最簡單的請求-應(yīng)答通信驗(yàn)證環(huán)境是否正常。這是服務(wù)端import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5555) while True: message socket.recv() print(f收到請求: {message.decode()}) socket.send(bworld)這是客戶端import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5555) socket.send(bhello) reply socket.recv() print(f收到回復(fù): {reply.decode()})先運(yùn)行服務(wù)端再運(yùn)行客戶端客戶端會打印出收到回復(fù): world。這批代碼有一個很值得品味的點(diǎn)客戶端connect后并沒有做任何握手邏輯服務(wù)端也沒有accept。ZMQ把連接過程完全封裝了你只管收發(fā)消息。我第一次跑通時確實(shí)有種這也太省事了吧的感嘆。2. 請求-應(yīng)答模式REQ/REP的嚴(yán)格一問一答2.1 REQ/REP的工作機(jī)制與鎖死現(xiàn)象請求-應(yīng)答模式是最容易上手的模式它模擬的是函數(shù)調(diào)用的語義客戶端發(fā)一條請求服務(wù)端處理完后必須回一條響應(yīng)一問一答節(jié)奏非常嚴(yán)格。REQ和REP的嚴(yán)格性體現(xiàn)在它們各自的狀態(tài)機(jī)上。REQ socket在send之后不能再次send必須等recv拿到響應(yīng)后才能進(jìn)入下一輪發(fā)送。REP socket在recv到請求后不能再次recv必須send返回響應(yīng)后才能接收下一條請求。這個設(shè)計(jì)保證了消息不會交錯但也帶來一個新手最常踩的坑——鎖死。我見過很多人在REQ端連發(fā)兩條消息代碼大概是這樣的socket.send(brequest-1) # 忘記 recv socket.send(brequest-2)程序會直接拋異?；蛘呖ㄋ?。REQ在第一次send后狀態(tài)變?yōu)榈却憫?yīng)此時再send就違反了狀態(tài)機(jī)。另一種鎖死情況是REP端處理完請求后忘了send響應(yīng)那REQ端會一直阻塞在recv。所以使用REQ/REP時一定要記住REQ的send必然對應(yīng)REP的recvREP的send必然對應(yīng)REQ的recv這是一條嚴(yán)格閉環(huán)。2.2 代碼實(shí)測同步請求的服務(wù)端與客戶端下面是更豐富一點(diǎn)的示例模擬一個加法計(jì)算服務(wù)。服務(wù)端接收兩個整數(shù)返回它們的和。服務(wù)端代碼import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5556) print(加法服務(wù)已啟動監(jiān)聽端口 5556) while True: # 接收請求格式為 bnum1 num2 request socket.recv() text request.decode() parts text.split() if len(parts) ! 2: socket.send(binvalid request) continue try: a, b int(parts[0]), int(parts[1]) result a b socket.send(str(result).encode()) except ValueError: socket.send(binvalid number)客戶端代碼import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5556) for i in range(10): socket.send(f{i} {i * 2}.encode()) reply socket.recv() print(f請求: {i} {i * 2}響應(yīng): {reply.decode()})運(yùn)行后客戶端會順序打印10次計(jì)算結(jié)果。注意這里每次send后必須等recv完成服務(wù)端處理慢時客戶端就會等待這正是同步調(diào)用的一種體現(xiàn)。2.3 REQ/REP逃不出的限制與擴(kuò)展思路REQ/REP簡單可靠但限制也很明顯同步、阻塞、一對一。如果服務(wù)端處理一條請求需要2秒那這個服務(wù)端同一時間就只能服務(wù)一個客戶端其他客戶端的請求全部排隊(duì)。面對并發(fā)場景很多人第一反應(yīng)是給客戶端開多線程每個線程獨(dú)立REQ socket。這個做法有效但線程數(shù)量一多連接數(shù)也會暴增并不優(yōu)雅。更合適的做法是引入ZMQ的ROUTER和DEALER socketROUTER接收客戶端請求發(fā)給一群DEALER worker處理worker把結(jié)果傳回ROUTER再轉(zhuǎn)給對應(yīng)客戶端。這就是經(jīng)典的經(jīng)紀(jì)人模式。不過ROUTER/DEALER屬于進(jìn)階內(nèi)容本篇先點(diǎn)到為止下篇會專門展開。先記住結(jié)論如果你的REQ/REP遇到了性能瓶頸別在REP端堆線程要考慮升級成ROUTER/DEALER架構(gòu)。3. 發(fā)布-訂閱模式PUB/SUB的廣播世界3.1 PUB/SUB的消息流向與訂閱過濾機(jī)制發(fā)布-訂閱模式解決的是一對多廣播問題和REQ/REP完全不同發(fā)布者PUB只負(fù)責(zé)發(fā)不知道也不關(guān)心消息有沒有人收到訂閱者SUB只負(fù)責(zé)收不需要給發(fā)布者任何應(yīng)答。這里有一個特別反直覺的設(shè)計(jì)SUB端必須設(shè)置訂閱過濾器才能收到消息。如果你只創(chuàng)建一個SUB并connect不設(shè)置任何過濾條件就recv程序會一直卡著收不到任何數(shù)據(jù)。原因在于ZMQ的設(shè)計(jì)者把是否訂閱所有消息這個選擇權(quán)交給了SUB端。設(shè)置過濾器的方法是在SUB socket上調(diào)用setsockopt(zmq.SUBSCRIBE, b)空字節(jié)串表示訂閱所有消息。也可以傳前綴比如bweather表示只接收以weather開頭的消息。這個消息過濾是在哪里發(fā)生的在PUB端。SUB端即使設(shè)了過濾器也不會把過濾條件傳給所有下游而是在建立連接時把自己的訂閱信息告訴PUB。PUB會根據(jù)所有訂閱者的過濾器決定是否把消息塞進(jìn)對應(yīng)的連接隊(duì)列。這樣做的好處是節(jié)省了網(wǎng)絡(luò)帶寬——不相關(guān)的消息根本不會發(fā)出。3.2 代碼實(shí)測天氣播報的發(fā)布訂閱寫一個天氣播報的例子。發(fā)布者持續(xù)發(fā)布包含城市和溫度的天氣預(yù)報消息訂閱者分為兩類一類訂閱所有城市的消息另一類只訂閱Shanghai的消息。先看發(fā)布端import zmq import time import random context zmq.Context() socket context.socket(zmq.PUB) socket.bind(tcp://*:5557) print(天氣發(fā)布服務(wù)已啟動) cities [Beijing, Shanghai, Guangzhou, Shenzhen] while True: city random.choice(cities) temp random.randint(-5, 35) message f{city} {temp}度.encode() socket.send(message) print(f發(fā)布: {message.decode()}) time.sleep(1)訂閱端訂閱全部信息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, b) # 訂閱所有消息 while True: message socket.recv() print(f收到: {message.decode()})另一個訂閱端只訂閱Shanghai的消息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, bShanghai) # 只訂閱Shanghai前綴 while True: message socket.recv() print(f上海天氣: {message.decode()})運(yùn)行發(fā)布端再運(yùn)行任意訂閱端你會看到訂閱端穩(wěn)定收到消息。注意順序必須先啟動發(fā)布端再啟動訂閱端否則訂閱端剛啟動時可能錯過前幾條消息。這是因?yàn)镾UB連接PUB需要時間在連接建立完成前PUB發(fā)出的消息會被丟棄。3.3 慢消費(fèi)者與消息丟棄PUB/SUB最常踩的坑PUB/SUB最經(jīng)典的坑有兩個。第一個是慢消費(fèi)者問題。假設(shè)PUB每秒發(fā)1000條消息某個SUB處理能力只有每秒100條它處理不過來。ZMQ的處理策略不是無限等待而是設(shè)置了高水位HWMHigh Water Mark。當(dāng)某個SUB的接收隊(duì)列堆積超過HWM值默認(rèn)1000條時PUB會直接丟棄該訂閱者的溢出消息。這個特性在很多場景下是好事——它避免了某個慢消費(fèi)者拖垮整個發(fā)布端。但如果你恰好需要消息必須全部送達(dá)PUB/SUB就不能滿足要求你需要更可靠的消息系統(tǒng)或者自己設(shè)計(jì)ACK機(jī)制。我在實(shí)際項(xiàng)目里處理這個問題的做法是給重要消息加序號訂閱端檢測到序號跳變時主動向發(fā)布端請求補(bǔ)償這個模式其實(shí)就是可靠發(fā)布訂閱的簡化版工作得很好。第二個坑是訂閱者遲到。在ZMQ里如果SUB先啟動、PUB后啟動SUB將收不到PUB啟動之前的所有消息。它不像Kafka那樣有持久化日志PUB發(fā)出的消息如果當(dāng)時沒有訂閱者連接消息就直接消失了。所以如果你需要一個消息必須被消費(fèi)的業(yè)務(wù)場景PUB/SUB從一開始就不該是首選。4. 推拉模式PUSH/PULL的數(shù)據(jù)管道4.1 任務(wù)分發(fā)的公平調(diào)度原理推拉模式PUSH/PULL最初是為處理管道設(shè)計(jì)的上游PUSH把任務(wù)推給下游PULLPULL處理完后再把結(jié)果PUSH給下一個節(jié)點(diǎn)。它和REQ/REP的區(qū)別在于PUSH端根本不在乎有沒有人接收處理只管推送和PUB/SUB的區(qū)別在于每條消息只給一個PULL接收者而不是廣播給所有人。當(dāng)多個PULL節(jié)點(diǎn)連接同一個PUSH時PUSH采用的是公平排隊(duì)round-robin策略它維護(hù)一個接收者列表依次把消息輪流發(fā)送給每個接收者。如果其中一個PULL節(jié)點(diǎn)斷開它會自動從列表移除之后消息就只分給仍然在線的節(jié)點(diǎn)。這個公平分發(fā)特性非常適合做并行任務(wù)分發(fā)。我在數(shù)據(jù)采集場景里用過這種結(jié)構(gòu)一個任務(wù)源PUSH把成千上萬個URL推給多個采集worker所在的PULL節(jié)點(diǎn)每個worker拿到的URL數(shù)量大致相同采集完成后各自處理。4.2 代碼實(shí)測三階段處理管道用一個三階段管道來演示。第一個進(jìn)程是任務(wù)源生成100個任務(wù)并PUSH出去第二個進(jìn)程是處理節(jié)點(diǎn)PULL任務(wù)、把字符串轉(zhuǎn)大寫、再PUSH到下一個階段第三個進(jìn)程是結(jié)果收尾PULL并打印結(jié)果。先看任務(wù)源import zmq context zmq.Context() socket context.socket(zmq.PUSH) socket.bind(tcp://*:5558) print(任務(wù)源啟動開始推送任務(wù)) for i in range(100): task ftask-{i}.encode() socket.send(task) if i % 20 0: print(f已推送: {task.decode()})處理節(jié)點(diǎn)這里我建議你可以啟動兩個實(shí)例來驗(yàn)證負(fù)載均衡import zmq import time context zmq.Context() receiver context.socket(zmq.PULL) receiver.connect(tcp://localhost:5558) sender context.socket(zmq.PUSH) sender.bind(tcp://*:5559) print(處理節(jié)點(diǎn)啟動等待任務(wù)...) while True: task receiver.recv() processed task.decode().upper().encode() sender.send(processed) print(f處理完成: {processed.decode()}) time.sleep(0.01) # 模擬處理耗時任務(wù)源和處理節(jié)點(diǎn)可以放在同一臺機(jī)器上但ZMQ在這種跨端口的連接場景下到底誰bind誰connect其實(shí)完全看設(shè)計(jì)。上面示例中任務(wù)源bind了5558處理節(jié)點(diǎn)connect過去再bind自己的輸出端口5559。你可以開兩個處理節(jié)點(diǎn)進(jìn)程它們會分別從5558拿任務(wù)兩個進(jìn)程之間呈現(xiàn)輪流拿任務(wù)的現(xiàn)象這正好驗(yàn)證了round-robin分發(fā)。結(jié)果收尾進(jìn)程import zmq context zmq.Context() socket context.socket(zmq.PULL) socket.connect(tcp://localhost:5559) print(結(jié)果收集器啟動) count 0 while count 100: message socket.recv() print(f最終結(jié)果: {message.decode()}) count 1先啟動結(jié)果收尾再啟動處理節(jié)點(diǎn)最后啟動任務(wù)源就能看到一個完整的管道跑通。如果把任務(wù)源的消息量增大你還會發(fā)現(xiàn)某個處理節(jié)點(diǎn)偶爾會比另一個收到的任務(wù)更多這取決于ZMQ內(nèi)部對每個連接的排隊(duì)情況但總體趨勢是均衡的。4.3 管道模式的邊界背壓與可靠性PUSH/PULL適合做數(shù)據(jù)管道但它同樣有邊界條件。第一個是背壓問題如果下游處理速度遠(yuǎn)慢于上游推送速度PUSH端的發(fā)送隊(duì)列會持續(xù)膨脹。ZMQ在達(dá)到HWM后PUSH端的send會進(jìn)入阻塞狀態(tài)這算是ZMQ少有的天然限流機(jī)制。阻塞帶來的連鎖反應(yīng)是上游任務(wù)源也會變慢但至少不會丟消息。相比之下PUB/SUB是直接丟棄PUSH/PULL是阻塞等待這是兩種截然不同的行為差異。第二個是可靠性問題。PUSH/PULL不保證消息傳輸可靠性。如果某個PULL節(jié)點(diǎn)在收到消息后、處理完成前崩潰了那條消息就丟失了。解決思路通常是在業(yè)務(wù)數(shù)據(jù)里增加任務(wù)編號配合一個存儲層做冪等控制或者使用可靠消息中間件來兜底。ZMQ本身是智能的傳輸層不是一個自帶分布式事務(wù)的消息平臺。5. 三種模式怎么選從業(yè)務(wù)場景倒推通信方案5.1 一張對照表看清三者的差異了解單個模式之后再來看整體我梳理了一張對照表直接從業(yè)務(wù)場景倒推通信方案參考價值很高。維度REQ/REPPUB/SUBPUSH/PULL消息流向雙向嚴(yán)格一問一答單向一對多廣播單向一對一分發(fā)發(fā)送方狀態(tài)必須等待應(yīng)答不關(guān)心誰接收不關(guān)心接收結(jié)果接收方數(shù)量通常一對一多個訂閱者多個接收者消息是否可能丟失不會阻塞等應(yīng)答會慢消費(fèi)者和遲到訂閱者會丟可能節(jié)點(diǎn)崩潰則消息消失典型場景遠(yuǎn)程函數(shù)調(diào)用、命令下發(fā)狀態(tài)廣播、日志分發(fā)、行情推送任務(wù)分發(fā)、流水線處理負(fù)載均衡能力弱單服務(wù)端串行處理無所有訂閱者都收同一份強(qiáng)按隊(duì)列均衡分配這張表里最值得關(guān)注的是消息是否可能丟失這一行。在實(shí)際生產(chǎn)環(huán)境里這是選型的第一決定因素。我自己的經(jīng)驗(yàn)是能容忍丟失且希望多接收方的用PUB/SUB需要嚴(yán)格同步響應(yīng)的用REQ/REP需要任務(wù)分流且可容忍極少丟失的用PUSH/PULL。5.2 混合使用一個服務(wù)同時扮演多個角色前面講的是三個獨(dú)立模式但真實(shí)世界里服務(wù)往往不是只用一種socket的。ZMQ允許你在同一個進(jìn)程里創(chuàng)建多個socket分別扮演不同角色這簡直是為復(fù)雜的分布式系統(tǒng)量身定制的。舉個例子一個采集服務(wù)可以同時做三件事用REP master接口接收管理者下發(fā)的采集任務(wù)用PUSH把URL推給下游解析worker再用SUB訂閱一個全局狀態(tài)頻道接收暫停/恢復(fù)指令。三個socket互不干擾協(xié)調(diào)工作。這種組合能力是ZMQ相比很多消息中間件更靈活的地方也是我后來越來越喜歡它的原因?；旌鲜褂脮r要注意一件事socket類型和模式必須嚴(yán)格匹配不能拿一個REP去連另一個REP也不能讓PUSH端去recv。ZMQ會在消息層面主動拒絕不兼容的模式配對報錯信息也足夠明確遇到報錯時先檢查一下兩端socket類型是否匹配大部分問題都能快速定位。5.3 上篇小結(jié)與后續(xù)方向預(yù)告寫到這里ZMQ的三種基礎(chǔ)通信模式應(yīng)該已經(jīng)很清晰了。REQ/REP是一問一答的同步調(diào)用PUB/SUB是廣播式的消息分發(fā)PUSH/PULL是均衡式的數(shù)據(jù)管道。它們覆蓋了分布式系統(tǒng)中絕大多數(shù)的通信需求而且代碼量極低對比手寫socket或者部署一個完整消息中間件性價比非常高。我個人在實(shí)際使用中最大的體會是用ZMQ一定不要慣性思維。很多從HTTP或者傳統(tǒng)消息隊(duì)列轉(zhuǎn)過來的開發(fā)者容易下意識地去找服務(wù)端客戶端請求地址這些概念但ZMQ的模式思維是角色決定行為——同一臺機(jī)器、同一個端口可以在一套架構(gòu)里既是服務(wù)端又是客戶端。想通這一點(diǎn)ZMQ的世界就會豁然開朗。還有一點(diǎn)值得提在跑示例時如果發(fā)現(xiàn)消息延遲或丟失先檢查是不是忘了設(shè)置SUBSCRIBE過濾器再檢查HWM和LINGER設(shè)置。HWM決定隊(duì)列上限LINGER影響socket關(guān)閉時是否等待未發(fā)送消息這兩個都是ZMQ調(diào)優(yōu)里的高頻參數(shù)。ZMQ的進(jìn)階武器還沒講完面向多對多路由的ROUTER/DEALER、進(jìn)程內(nèi)通信的inproc優(yōu)化、發(fā)布訂閱的可靠代理模式、多消息幀的處理技巧這些內(nèi)容都留到下篇再和各位細(xì)聊。每一種模式背后都有非常實(shí)用的生產(chǎn)場景把基礎(chǔ)打牢之后進(jìn)階內(nèi)容會順暢很多。 ZMQ三種通信模式詳解：REQ/REP、PUB/SUB、PUSH/PULL實(shí)戰(zhàn)
最近在做一個分布式采集的中間件改造幾個服務(wù)之間需要頻繁傳遞任務(wù)消息最開始圖省事直接上了HTTP接口輪詢結(jié)果超時重試、連接風(fēng)暴、數(shù)據(jù)丟失輪著來。后來換成了ZMQZeroMQ這套消息通信庫才算把這塊理順。這個系列打算把我對ZMQ通信模式的理解完整梳理一遍本篇是上篇重點(diǎn)講最常用的三種基礎(chǔ)模式請求-應(yīng)答REQ/REP、發(fā)布-訂閱PUB/SUB、推拉管道PUSH/PULL以及它們各自適合什么場景、有哪些坑配合可運(yùn)行的代碼示例方便你直接照著復(fù)現(xiàn)。如果你正在做微服務(wù)通信、分布式任務(wù)調(diào)度、數(shù)據(jù)采集管道這類東西又不想引入Kafka或RabbitMQ這種重量級中間件那ZMQ會是一個非常合適的選擇。它足夠輕沒有獨(dú)立Broker進(jìn)程功能又足夠硬核理解好通信模式就等于拿到了它的使用鑰匙。1. 從裸socket到消息總線的跳躍ZMQ到底解決了什么問題1.1 傳統(tǒng)socket開發(fā)中最讓人頭疼的三件事很多人剛接觸ZMQ時容易把它當(dāng)成一個普通的網(wǎng)絡(luò)庫但實(shí)際上它解決的是如何優(yōu)雅地建立分布式消息通道這個層面的問題。用裸socket寫通信時你大概率會被三件事折磨。第一是連接管理。兩個進(jìn)程要通信你得手動處理建立連接、斷開重連、心跳?；睢＿B接斷了他不會自己恢復(fù)你需要寫一堆重試邏輯。第二是消息邊界。TCP是字節(jié)流你發(fā)了hello和world對端收到的可能是helloworld、也可能是hel和loworld你得自己設(shè)計(jì)報文頭、長度字段、粘包拆包邏輯。第三是架構(gòu)靈活性。裸socket天然是1對1的想要1對多廣播、多對多消息路由那得自己實(shí)現(xiàn)一套注冊發(fā)現(xiàn)機(jī)制。ZMQ把這三件事全部收納了。它內(nèi)置了斷線重連機(jī)制你寫的代碼里根本不需要關(guān)心連接狀態(tài)——它自己會悄悄重連。它面向消息而非字節(jié)流每條消息有天然邊界不存在粘包拆包問題。它提供多種通信模式就像給消息傳輸定義了不同規(guī)矩你照著規(guī)矩連就行。一句話理解socket給你的是字節(jié)管道ZMQ給你的是消息總線后者自帶連接管理、消息分幀和多種路由策略。1.2 ZMQ的核心抽象消息、端點(diǎn)與上下文使用ZMQ需要先理解三個基礎(chǔ)對象Context、Socket和Endpoint。Context是ZMQ的總環(huán)境所有socket都必須在一個Context中創(chuàng)建類似進(jìn)程級的資源管理器。在pyzmq中通常一行代碼搞定ctx zmq.Context()。每個socket通過socket_type指定模式比如zmq.REQ、zmq.PUB。Endpoint是socket綁定的地址ZMQ支持多種傳輸協(xié)議最常用的是tcp://跨主機(jī)、ipc://同一主機(jī)進(jìn)程間、inproc://同一進(jìn)程內(nèi)線程間。這里要特別強(qiáng)調(diào)一個設(shè)計(jì)理念ZMQ的socket不像傳統(tǒng)socket那樣區(qū)分服務(wù)端和客戶端。一個socket可以bind綁定也可以connect連接而且規(guī)則很靈活——不是只有服務(wù)端才能bind客戶端也能bind服務(wù)端也能connect。理解這一點(diǎn)對后面的模式會很關(guān)鍵。1.3 環(huán)境準(zhǔn)備安裝與第一個連接動手之前先把環(huán)境裝上。Python環(huán)境用pip安裝pyzmqpip install pyzmq安裝完成后寫一個最簡單的請求-應(yīng)答通信驗(yàn)證環(huán)境是否正常。這是服務(wù)端import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5555) while True: message socket.recv() print(f收到請求: {message.decode()}) socket.send(bworld)這是客戶端import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5555) socket.send(bhello) reply socket.recv() print(f收到回復(fù): {reply.decode()})先運(yùn)行服務(wù)端再運(yùn)行客戶端客戶端會打印出收到回復(fù): world。這批代碼有一個很值得品味的點(diǎn)客戶端connect后并沒有做任何握手邏輯服務(wù)端也沒有accept。ZMQ把連接過程完全封裝了你只管收發(fā)消息。我第一次跑通時確實(shí)有種這也太省事了吧的感嘆。2. 請求-應(yīng)答模式REQ/REP的嚴(yán)格一問一答2.1 REQ/REP的工作機(jī)制與鎖死現(xiàn)象請求-應(yīng)答模式是最容易上手的模式它模擬的是函數(shù)調(diào)用的語義客戶端發(fā)一條請求服務(wù)端處理完后必須回一條響應(yīng)一問一答節(jié)奏非常嚴(yán)格。REQ和REP的嚴(yán)格性體現(xiàn)在它們各自的狀態(tài)機(jī)上。REQ socket在send之后不能再次send必須等recv拿到響應(yīng)后才能進(jìn)入下一輪發(fā)送。REP socket在recv到請求后不能再次recv必須send返回響應(yīng)后才能接收下一條請求。這個設(shè)計(jì)保證了消息不會交錯但也帶來一個新手最常踩的坑——鎖死。我見過很多人在REQ端連發(fā)兩條消息代碼大概是這樣的socket.send(brequest-1) # 忘記 recv socket.send(brequest-2)程序會直接拋異?；蛘呖ㄋ?。REQ在第一次send后狀態(tài)變?yōu)榈却憫?yīng)此時再send就違反了狀態(tài)機(jī)。另一種鎖死情況是REP端處理完請求后忘了send響應(yīng)那REQ端會一直阻塞在recv。所以使用REQ/REP時一定要記住REQ的send必然對應(yīng)REP的recvREP的send必然對應(yīng)REQ的recv這是一條嚴(yán)格閉環(huán)。2.2 代碼實(shí)測同步請求的服務(wù)端與客戶端下面是更豐富一點(diǎn)的示例模擬一個加法計(jì)算服務(wù)。服務(wù)端接收兩個整數(shù)返回它們的和。服務(wù)端代碼import zmq context zmq.Context() socket context.socket(zmq.REP) socket.bind(tcp://*:5556) print(加法服務(wù)已啟動監(jiān)聽端口 5556) while True: # 接收請求格式為 bnum1 num2 request socket.recv() text request.decode() parts text.split() if len(parts) ! 2: socket.send(binvalid request) continue try: a, b int(parts[0]), int(parts[1]) result a b socket.send(str(result).encode()) except ValueError: socket.send(binvalid number)客戶端代碼import zmq context zmq.Context() socket context.socket(zmq.REQ) socket.connect(tcp://localhost:5556) for i in range(10): socket.send(f{i} {i * 2}.encode()) reply socket.recv() print(f請求: {i} {i * 2}響應(yīng): {reply.decode()})運(yùn)行后客戶端會順序打印10次計(jì)算結(jié)果。注意這里每次send后必須等recv完成服務(wù)端處理慢時客戶端就會等待這正是同步調(diào)用的一種體現(xiàn)。2.3 REQ/REP逃不出的限制與擴(kuò)展思路REQ/REP簡單可靠但限制也很明顯同步、阻塞、一對一。如果服務(wù)端處理一條請求需要2秒那這個服務(wù)端同一時間就只能服務(wù)一個客戶端其他客戶端的請求全部排隊(duì)。面對并發(fā)場景很多人第一反應(yīng)是給客戶端開多線程每個線程獨(dú)立REQ socket。這個做法有效但線程數(shù)量一多連接數(shù)也會暴增并不優(yōu)雅。更合適的做法是引入ZMQ的ROUTER和DEALER socketROUTER接收客戶端請求發(fā)給一群DEALER worker處理worker把結(jié)果傳回ROUTER再轉(zhuǎn)給對應(yīng)客戶端。這就是經(jīng)典的經(jīng)紀(jì)人模式。不過ROUTER/DEALER屬于進(jìn)階內(nèi)容本篇先點(diǎn)到為止下篇會專門展開。先記住結(jié)論如果你的REQ/REP遇到了性能瓶頸別在REP端堆線程要考慮升級成ROUTER/DEALER架構(gòu)。3. 發(fā)布-訂閱模式PUB/SUB的廣播世界3.1 PUB/SUB的消息流向與訂閱過濾機(jī)制發(fā)布-訂閱模式解決的是一對多廣播問題和REQ/REP完全不同發(fā)布者PUB只負(fù)責(zé)發(fā)不知道也不關(guān)心消息有沒有人收到訂閱者SUB只負(fù)責(zé)收不需要給發(fā)布者任何應(yīng)答。這里有一個特別反直覺的設(shè)計(jì)SUB端必須設(shè)置訂閱過濾器才能收到消息。如果你只創(chuàng)建一個SUB并connect不設(shè)置任何過濾條件就recv程序會一直卡著收不到任何數(shù)據(jù)。原因在于ZMQ的設(shè)計(jì)者把是否訂閱所有消息這個選擇權(quán)交給了SUB端。設(shè)置過濾器的方法是在SUB socket上調(diào)用setsockopt(zmq.SUBSCRIBE, b)空字節(jié)串表示訂閱所有消息。也可以傳前綴比如bweather表示只接收以weather開頭的消息。這個消息過濾是在哪里發(fā)生的在PUB端。SUB端即使設(shè)了過濾器也不會把過濾條件傳給所有下游而是在建立連接時把自己的訂閱信息告訴PUB。PUB會根據(jù)所有訂閱者的過濾器決定是否把消息塞進(jìn)對應(yīng)的連接隊(duì)列。這樣做的好處是節(jié)省了網(wǎng)絡(luò)帶寬——不相關(guān)的消息根本不會發(fā)出。3.2 代碼實(shí)測天氣播報的發(fā)布訂閱寫一個天氣播報的例子。發(fā)布者持續(xù)發(fā)布包含城市和溫度的天氣預(yù)報消息訂閱者分為兩類一類訂閱所有城市的消息另一類只訂閱Shanghai的消息。先看發(fā)布端import zmq import time import random context zmq.Context() socket context.socket(zmq.PUB) socket.bind(tcp://*:5557) print(天氣發(fā)布服務(wù)已啟動) cities [Beijing, Shanghai, Guangzhou, Shenzhen] while True: city random.choice(cities) temp random.randint(-5, 35) message f{city} {temp}度.encode() socket.send(message) print(f發(fā)布: {message.decode()}) time.sleep(1)訂閱端訂閱全部信息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, b) # 訂閱所有消息 while True: message socket.recv() print(f收到: {message.decode()})另一個訂閱端只訂閱Shanghai的消息import zmq context zmq.Context() socket context.socket(zmq.SUB) socket.connect(tcp://localhost:5557) socket.setsockopt(zmq.SUBSCRIBE, bShanghai) # 只訂閱Shanghai前綴 while True: message socket.recv() print(f上海天氣: {message.decode()})運(yùn)行發(fā)布端再運(yùn)行任意訂閱端你會看到訂閱端穩(wěn)定收到消息。注意順序必須先啟動發(fā)布端再啟動訂閱端否則訂閱端剛啟動時可能錯過前幾條消息。這是因?yàn)镾UB連接PUB需要時間在連接建立完成前PUB發(fā)出的消息會被丟棄。3.3 慢消費(fèi)者與消息丟棄PUB/SUB最常踩的坑PUB/SUB最經(jīng)典的坑有兩個。第一個是慢消費(fèi)者問題。假設(shè)PUB每秒發(fā)1000條消息某個SUB處理能力只有每秒100條它處理不過來。ZMQ的處理策略不是無限等待而是設(shè)置了高水位HWMHigh Water Mark。當(dāng)某個SUB的接收隊(duì)列堆積超過HWM值默認(rèn)1000條時PUB會直接丟棄該訂閱者的溢出消息。這個特性在很多場景下是好事——它避免了某個慢消費(fèi)者拖垮整個發(fā)布端。但如果你恰好需要消息必須全部送達(dá)PUB/SUB就不能滿足要求你需要更可靠的消息系統(tǒng)或者自己設(shè)計(jì)ACK機(jī)制。我在實(shí)際項(xiàng)目里處理這個問題的做法是給重要消息加序號訂閱端檢測到序號跳變時主動向發(fā)布端請求補(bǔ)償這個模式其實(shí)就是可靠發(fā)布訂閱的簡化版工作得很好。第二個坑是訂閱者遲到。在ZMQ里如果SUB先啟動、PUB后啟動SUB將收不到PUB啟動之前的所有消息。它不像Kafka那樣有持久化日志PUB發(fā)出的消息如果當(dāng)時沒有訂閱者連接消息就直接消失了。所以如果你需要一個消息必須被消費(fèi)的業(yè)務(wù)場景PUB/SUB從一開始就不該是首選。4. 推拉模式PUSH/PULL的數(shù)據(jù)管道4.1 任務(wù)分發(fā)的公平調(diào)度原理推拉模式PUSH/PULL最初是為處理管道設(shè)計(jì)的上游PUSH把任務(wù)推給下游PULLPULL處理完后再把結(jié)果PUSH給下一個節(jié)點(diǎn)。它和REQ/REP的區(qū)別在于PUSH端根本不在乎有沒有人接收處理只管推送和PUB/SUB的區(qū)別在于每條消息只給一個PULL接收者而不是廣播給所有人。當(dāng)多個PULL節(jié)點(diǎn)連接同一個PUSH時PUSH采用的是公平排隊(duì)round-robin策略它維護(hù)一個接收者列表依次把消息輪流發(fā)送給每個接收者。如果其中一個PULL節(jié)點(diǎn)斷開它會自動從列表移除之后消息就只分給仍然在線的節(jié)點(diǎn)。這個公平分發(fā)特性非常適合做并行任務(wù)分發(fā)。我在數(shù)據(jù)采集場景里用過這種結(jié)構(gòu)一個任務(wù)源PUSH把成千上萬個URL推給多個采集worker所在的PULL節(jié)點(diǎn)每個worker拿到的URL數(shù)量大致相同采集完成后各自處理。4.2 代碼實(shí)測三階段處理管道用一個三階段管道來演示。第一個進(jìn)程是任務(wù)源生成100個任務(wù)并PUSH出去第二個進(jìn)程是處理節(jié)點(diǎn)PULL任務(wù)、把字符串轉(zhuǎn)大寫、再PUSH到下一個階段第三個進(jìn)程是結(jié)果收尾PULL并打印結(jié)果。先看任務(wù)源import zmq context zmq.Context() socket context.socket(zmq.PUSH) socket.bind(tcp://*:5558) print(任務(wù)源啟動開始推送任務(wù)) for i in range(100): task ftask-{i}.encode() socket.send(task) if i % 20 0: print(f已推送: {task.decode()})處理節(jié)點(diǎn)這里我建議你可以啟動兩個實(shí)例來驗(yàn)證負(fù)載均衡import zmq import time context zmq.Context() receiver context.socket(zmq.PULL) receiver.connect(tcp://localhost:5558) sender context.socket(zmq.PUSH) sender.bind(tcp://*:5559) print(處理節(jié)點(diǎn)啟動等待任務(wù)...) while True: task receiver.recv() processed task.decode().upper().encode() sender.send(processed) print(f處理完成: {processed.decode()}) time.sleep(0.01) # 模擬處理耗時任務(wù)源和處理節(jié)點(diǎn)可以放在同一臺機(jī)器上但ZMQ在這種跨端口的連接場景下到底誰bind誰connect其實(shí)完全看設(shè)計(jì)。上面示例中任務(wù)源bind了5558處理節(jié)點(diǎn)connect過去再bind自己的輸出端口5559。你可以開兩個處理節(jié)點(diǎn)進(jìn)程它們會分別從5558拿任務(wù)兩個進(jìn)程之間呈現(xiàn)輪流拿任務(wù)的現(xiàn)象這正好驗(yàn)證了round-robin分發(fā)。結(jié)果收尾進(jìn)程import zmq context zmq.Context() socket context.socket(zmq.PULL) socket.connect(tcp://localhost:5559) print(結(jié)果收集器啟動) count 0 while count 100: message socket.recv() print(f最終結(jié)果: {message.decode()}) count 1先啟動結(jié)果收尾再啟動處理節(jié)點(diǎn)最后啟動任務(wù)源就能看到一個完整的管道跑通。如果把任務(wù)源的消息量增大你還會發(fā)現(xiàn)某個處理節(jié)點(diǎn)偶爾會比另一個收到的任務(wù)更多這取決于ZMQ內(nèi)部對每個連接的排隊(duì)情況但總體趨勢是均衡的。4.3 管道模式的邊界背壓與可靠性PUSH/PULL適合做數(shù)據(jù)管道但它同樣有邊界條件。第一個是背壓問題如果下游處理速度遠(yuǎn)慢于上游推送速度PUSH端的發(fā)送隊(duì)列會持續(xù)膨脹。ZMQ在達(dá)到HWM后PUSH端的send會進(jìn)入阻塞狀態(tài)這算是ZMQ少有的天然限流機(jī)制。阻塞帶來的連鎖反應(yīng)是上游任務(wù)源也會變慢但至少不會丟消息。相比之下PUB/SUB是直接丟棄PUSH/PULL是阻塞等待這是兩種截然不同的行為差異。第二個是可靠性問題。PUSH/PULL不保證消息傳輸可靠性。如果某個PULL節(jié)點(diǎn)在收到消息后、處理完成前崩潰了那條消息就丟失了。解決思路通常是在業(yè)務(wù)數(shù)據(jù)里增加任務(wù)編號配合一個存儲層做冪等控制或者使用可靠消息中間件來兜底。ZMQ本身是智能的傳輸層不是一個自帶分布式事務(wù)的消息平臺。5. 三種模式怎么選從業(yè)務(wù)場景倒推通信方案5.1 一張對照表看清三者的差異了解單個模式之后再來看整體我梳理了一張對照表直接從業(yè)務(wù)場景倒推通信方案參考價值很高。維度REQ/REPPUB/SUBPUSH/PULL消息流向雙向嚴(yán)格一問一答單向一對多廣播單向一對一分發(fā)發(fā)送方狀態(tài)必須等待應(yīng)答不關(guān)心誰接收不關(guān)心接收結(jié)果接收方數(shù)量通常一對一多個訂閱者多個接收者消息是否可能丟失不會阻塞等應(yīng)答會慢消費(fèi)者和遲到訂閱者會丟可能節(jié)點(diǎn)崩潰則消息消失典型場景遠(yuǎn)程函數(shù)調(diào)用、命令下發(fā)狀態(tài)廣播、日志分發(fā)、行情推送任務(wù)分發(fā)、流水線處理負(fù)載均衡能力弱單服務(wù)端串行處理無所有訂閱者都收同一份強(qiáng)按隊(duì)列均衡分配這張表里最值得關(guān)注的是消息是否可能丟失這一行。在實(shí)際生產(chǎn)環(huán)境里這是選型的第一決定因素。我自己的經(jīng)驗(yàn)是能容忍丟失且希望多接收方的用PUB/SUB需要嚴(yán)格同步響應(yīng)的用REQ/REP需要任務(wù)分流且可容忍極少丟失的用PUSH/PULL。5.2 混合使用一個服務(wù)同時扮演多個角色前面講的是三個獨(dú)立模式但真實(shí)世界里服務(wù)往往不是只用一種socket的。ZMQ允許你在同一個進(jìn)程里創(chuàng)建多個socket分別扮演不同角色這簡直是為復(fù)雜的分布式系統(tǒng)量身定制的。舉個例子一個采集服務(wù)可以同時做三件事用REP master接口接收管理者下發(fā)的采集任務(wù)用PUSH把URL推給下游解析worker再用SUB訂閱一個全局狀態(tài)頻道接收暫停/恢復(fù)指令。三個socket互不干擾協(xié)調(diào)工作。這種組合能力是ZMQ相比很多消息中間件更靈活的地方也是我后來越來越喜歡它的原因?；旌鲜褂脮r要注意一件事socket類型和模式必須嚴(yán)格匹配不能拿一個REP去連另一個REP也不能讓PUSH端去recv。ZMQ會在消息層面主動拒絕不兼容的模式配對報錯信息也足夠明確遇到報錯時先檢查一下兩端socket類型是否匹配大部分問題都能快速定位。5.3 上篇小結(jié)與后續(xù)方向預(yù)告寫到這里ZMQ的三種基礎(chǔ)通信模式應(yīng)該已經(jīng)很清晰了。REQ/REP是一問一答的同步調(diào)用PUB/SUB是廣播式的消息分發(fā)PUSH/PULL是均衡式的數(shù)據(jù)管道。它們覆蓋了分布式系統(tǒng)中絕大多數(shù)的通信需求而且代碼量極低對比手寫socket或者部署一個完整消息中間件性價比非常高。我個人在實(shí)際使用中最大的體會是用ZMQ一定不要慣性思維。很多從HTTP或者傳統(tǒng)消息隊(duì)列轉(zhuǎn)過來的開發(fā)者容易下意識地去找服務(wù)端客戶端請求地址這些概念但ZMQ的模式思維是角色決定行為——同一臺機(jī)器、同一個端口可以在一套架構(gòu)里既是服務(wù)端又是客戶端。想通這一點(diǎn)ZMQ的世界就會豁然開朗。還有一點(diǎn)值得提在跑示例時如果發(fā)現(xiàn)消息延遲或丟失先檢查是不是忘了設(shè)置SUBSCRIBE過濾器再檢查HWM和LINGER設(shè)置。HWM決定隊(duì)列上限LINGER影響socket關(guān)閉時是否等待未發(fā)送消息這兩個都是ZMQ調(diào)優(yōu)里的高頻參數(shù)。ZMQ的進(jìn)階武器還沒講完面向多對多路由的ROUTER/DEALER、進(jìn)程內(nèi)通信的inproc優(yōu)化、發(fā)布訂閱的可靠代理模式、多消息幀的處理技巧這些內(nèi)容都留到下篇再和各位細(xì)聊。每一種模式背后都有非常實(shí)用的生產(chǎn)場景把基礎(chǔ)打牢之后進(jìn)階內(nèi)容會順暢很多。