簡介這份資源面向在 Windows 平臺使用 VS2008 進(jìn)行 ZeroMQ 開發(fā)的初學(xué)者圍繞訂閱與發(fā)布模式提供一套可快速上手的入門實(shí)例。包內(nèi)包含完整的服務(wù)端與客戶端測試工程以及一份 Windows 下 VS2008 使用 ZeroMQ 的說明文檔幫助讀者理解如何編譯 ZeroMQ 庫、配置頭文件與庫文件依賴并搭建自己的發(fā)布訂閱通信程序。資源共 31 個(gè)文件以 h 頭文件、cpp 源文件、dll 動態(tài)庫、lib 導(dǎo)入庫為主另有 vcproj 工程文件、sln 解決方案及若干說明文本壓縮包約 4.25MB目錄結(jié)構(gòu)清晰便于按模塊查閱。目前已有 488 人學(xué)習(xí)下載。通過這份資料讀者可以省去從零摸索編譯配置的時(shí)間直接對照工程結(jié)構(gòu)完成環(huán)境搭建掌握 ZeroMQ 在 VS2008 下的基本集成方式與訂閱發(fā)布通信流程適合需要快速入門消息隊(duì)列開發(fā)的讀者參考。1. 老 IDE 配新消息庫為什么這套組合還在被翻出來用VS2008 配 ZeroMQ聽起來像從技術(shù)博物館里翻出來的組合。但如果你手上維護(hù)的是工控上位機(jī)、醫(yī)療設(shè)備采集端或者某類產(chǎn)線測試軟件這個(gè)組合一點(diǎn)都不陌生——這些場景里 XP 嵌入式或 Win7 工控機(jī)還在跑VS2008 是唯一能出穩(wěn)定 Release 包的編譯環(huán)境而 ZeroMQ 的發(fā)布訂閱模式恰好能解決多臺設(shè)備狀態(tài)廣播的問題。問題在于ZeroMQ 官方早已不提供 VS2008 的工程文件網(wǎng)上能搜到的教程要么是 VS2010 以后的要么只給源碼不給編譯步驟。這份資源的價(jià)值就在于它把從源碼編譯到 pub/sub 實(shí)例跑通的完整鏈路補(bǔ)上了包括那個(gè)年代特有的 CRT 版本沖突和 socket 庫鏈接順序問題。適合兩類人一是被老項(xiàng)目綁住、必須用 VS2008 出包的一線維護(hù)者二是想理解 ZeroMQ 發(fā)布訂閱底層行為、不依賴高級封裝的開發(fā)者。2. 編譯 ZeroMQ 源碼從 libzmq 到 VS2008 可用的靜態(tài)庫2.1 為什么不用官方預(yù)編譯包ZeroMQ 官方在 4.x 之后就不再提供 VS2008 的二進(jìn)制包了NuGet 上的 libzmq 包最低要求 VS2010 的 MSBuild 工具集。直接拿 VS2010 編譯的 lib 往 VS2008 工程里鏈大概率會在鏈接階段報(bào)LNK2001: 無法解析的外部符號 _zmq_ctx_new因?yàn)閮蓚€(gè)版本的 CRT 對符號修飾規(guī)則不同。常見做法是下載 libzmq 的源碼包用 VS2008 自帶的 vcvarsall.bat 環(huán)境重新編譯。我一般會選 libzmq 4.1.x 這個(gè)分支它對 C98 的兼容性最好再往后的版本用了不少 C11 特性VS2008 的編譯器認(rèn)不全。源碼目錄里有個(gè)builds/msvc文件夾里面是給 VS2010 以上用的 vcxproj 文件。VS2008 打不開這些工程需要手動建一個(gè) Win32 靜態(tài)庫工程把src目錄下的 .cpp 和 .h 文件全部加進(jìn)去。注意要排除src/curve_client.cpp和src/curve_server.cpp這兩個(gè)文件它們依賴 libsodium而 libsodium 在 VS2008 下編譯更麻煩普通 pub/sub 場景用不到加密。2.2 手動建工程的參數(shù)配置在 VS2008 里新建 Win32 項(xiàng)目類型選“靜態(tài)庫”不要勾“預(yù)編譯頭”。然后把 libzmq 源碼include目錄和src目錄下的文件按以下規(guī)則加入# 目錄結(jié)構(gòu)參考假設(shè)源碼解壓在 D:\libzmq-4.1.7 D:\libzmq-4.1.7\include\zmq.h D:\libzmq-4.1.7\include\zmq_utils.h D:\libzmq-4.1.7\src\*.cpp # 排除 curve_client.cpp 和 curve_server.cpp D:\libzmq-4.1.7\src\*.hpp項(xiàng)目屬性里需要改三個(gè)地方。第一C/C → 預(yù)處理器 → 預(yù)處理器定義加上ZMQ_STATIC和_CRT_SECURE_NO_WARNINGS。ZMQ_STATIC告訴頭文件不要用__declspec(dllimport)修飾函數(shù)聲明否則鏈接時(shí)會去找 DLL 里的符號。第二C/C → 代碼生成 → 運(yùn)行時(shí)庫選“多線程 (/MT)”不要選 DLL 版本的 /MD否則 Release 包拿到?jīng)]裝 VC 運(yùn)行庫的工控機(jī)上會直接崩。第三鏈接器 → 輸入 → 附加依賴項(xiàng)加上ws2_32.lib和Iphlpapi.libZeroMQ 內(nèi)部用到了 Windows 的 socket 和 IP 輔助函數(shù)。編譯時(shí)如果報(bào)error C2065: “sockaddr_in”: 未聲明的標(biāo)識符說明winsock2.h的包含順序有問題。libzmq 的源碼里有些文件先包含了windows.h而windows.h默認(rèn)會拉進(jìn)winsock.h和winsock2.h沖突。解決辦法是在項(xiàng)目屬性 C/C → 預(yù)處理器里加WIN32_LEAN_AND_MEAN這個(gè)宏會讓windows.h跳過 winsock 部分。編譯成功后會在Release目錄下生成libzmq.lib大約 2MB 左右。這個(gè) lib 就是后面所有實(shí)例的基礎(chǔ)。2.3 驗(yàn)證靜態(tài)庫是否可用不要急著寫 pub/sub先寫一個(gè)最簡單的zmq_version調(diào)用驗(yàn)證鏈接是否正常// test_zmq.cpp : 驗(yàn)證 libzmq 靜態(tài)庫鏈接 #include zmq.h #include stdio.h int main() { int major, minor, patch; zmq_version(major, minor, patch); printf(ZeroMQ version: %d.%d.%d\n, major, minor, patch); return 0; }在 VS2008 里新建 Win32 控制臺工程把這個(gè)文件加進(jìn)去鏈接器附加依賴項(xiàng)填上剛才編譯出的libzmq.lib的完整路徑。運(yùn)行后如果輸出ZeroMQ version: 4.1.7說明靜態(tài)庫沒問題。如果報(bào)LNK2019: 無法解析的外部符號 _zmq_version檢查兩點(diǎn)一是ZMQ_STATIC宏有沒有加二是 lib 文件的路徑有沒有寫對。這一步過了后面的 pub/sub 才有意義。3. 發(fā)布訂閱實(shí)例從單進(jìn)程到跨進(jìn)程的完整代碼3.1 發(fā)布端代碼與參數(shù)說明發(fā)布端創(chuàng)建一個(gè)ZMQ_PUB類型的 socket綁定到某個(gè)端口然后循環(huán)發(fā)送消息。下面這段代碼可以直接在 VS2008 里編譯運(yùn)行// publisher.cpp : ZeroMQ 發(fā)布端 #include zmq.h #include stdio.h #include string.h #include windows.h int main() { void* context zmq_ctx_new(); void* publisher zmq_socket(context, ZMQ_PUB); // 綁定到 5555 端口所有網(wǎng)卡都可訪問 int rc zmq_bind(publisher, tcp://*:5555); if (rc ! 0) { printf(bind failed: %s\n, zmq_strerror(zmq_errno())); return 1; } int count 0; while (1) { char buffer[64]; // 消息格式主題 空格 序號 sprintf(buffer, status %d, count); // 發(fā)送消息第三個(gè)參數(shù)是長度第四個(gè)參數(shù) 0 表示阻塞發(fā)送 rc zmq_send(publisher, buffer, strlen(buffer), 0); if (rc 0) { printf(send failed: %s\n, zmq_strerror(zmq_errno())); break; } printf(sent: %s\n, buffer); Sleep(1000); // 每秒發(fā)一條 } zmq_close(publisher); zmq_ctx_destroy(context); return 0; }zmq_socket的第二個(gè)參數(shù)ZMQ_PUB決定了這個(gè) socket 只能發(fā)送不能接收。zmq_bind里的tcp://*:5555表示監(jiān)聽本機(jī)所有網(wǎng)卡的 5555 端口如果只想本機(jī)測試可以改成tcp://127.0.0.1:5555。zmq_send的第四個(gè)參數(shù)是標(biāo)志位0 表示阻塞模式如果發(fā)送緩沖區(qū)滿了會等待如果不想等可以傳ZMQ_DONTWAIT但 pub 模式下一般不會滿因?yàn)?pub 端不維護(hù)訂閱者狀態(tài)發(fā)出去就不管了。這里有個(gè)容易忽略的點(diǎn)Sleep(1000)用的是 Windows API需要包含windows.h。如果不想依賴 Windows 頭文件可以用zmq_sleep(1)ZeroMQ 自己封裝了一個(gè)跨平臺的 sleep 函數(shù)參數(shù)單位是秒。3.2 訂閱端代碼與過濾機(jī)制訂閱端創(chuàng)建一個(gè)ZMQ_SUB類型的 socket連接到發(fā)布端地址然后通過zmq_setsockopt設(shè)置訂閱過濾前綴// subscriber.cpp : ZeroMQ 訂閱端 #include zmq.h #include stdio.h #include string.h int main() { void* context zmq_ctx_new(); void* subscriber zmq_socket(context, ZMQ_SUB); // 連接到發(fā)布端 int rc zmq_connect(subscriber, tcp://127.0.0.1:5555); if (rc ! 0) { printf(connect failed: %s\n, zmq_strerror(zmq_errno())); return 1; } // 設(shè)置訂閱過濾前綴空字符串表示接收所有消息 const char* filter status; rc zmq_setsockopt(subscriber, ZMQ_SUBSCRIBE, filter, strlen(filter)); if (rc ! 0) { printf(setsockopt failed: %s\n, zmq_strerror(zmq_errno())); return 1; } while (1) { char buffer[64]; // 接收消息阻塞模式 int size zmq_recv(subscriber, buffer, sizeof(buffer) - 1, 0); if (size 0) { printf(recv failed: %s\n, zmq_strerror(zmq_errno())); break; } buffer[size] \0; printf(received: %s\n, buffer); } zmq_close(subscriber); zmq_ctx_destroy(context); return 0; }zmq_setsockopt的第三個(gè)參數(shù)ZMQ_SUBSCRIBE是訂閱選項(xiàng)第四個(gè)參數(shù)是過濾前綴。ZeroMQ 的過濾是在發(fā)布端做的訂閱端設(shè)置的前綴會通過底層協(xié)議傳到發(fā)布端發(fā)布端只發(fā)送匹配前綴的消息。這意味著如果訂閱端設(shè)置的前綴是status那么發(fā)布端發(fā)送sensor 1時(shí)訂閱端根本收不到消息在發(fā)布端就被丟棄了。這個(gè)機(jī)制的好處是節(jié)省網(wǎng)絡(luò)帶寬壞處是發(fā)布端需要維護(hù)每個(gè)訂閱端的過濾規(guī)則訂閱端數(shù)量多的時(shí)候發(fā)布端內(nèi)存會漲。zmq_recv的第四個(gè)參數(shù) 0 表示阻塞接收如果沒有消息會一直等。如果不想阻塞可以傳ZMQ_DONTWAIT但返回 -1 時(shí)需要判斷zmq_errno()是不是EAGAIN是的話表示暫時(shí)沒消息不是錯(cuò)誤。3.3 編譯運(yùn)行與跨進(jìn)程驗(yàn)證把 publisher.cpp 和 subscriber.cpp 分別建成兩個(gè) Win32 控制臺工程鏈接同一個(gè)libzmq.lib。先運(yùn)行 publisher再運(yùn)行 subscribersubscriber 的控制臺會每秒打印一條received: status N。如果 subscriber 先啟動、publisher 后啟動subscriber 也能收到消息因?yàn)?ZeroMQ 的 connect 是異步的會自動重連。跨進(jìn)程驗(yàn)證時(shí)注意防火墻。Windows 防火墻默認(rèn)會攔截沒有數(shù)字簽名的程序監(jiān)聽端口第一次運(yùn)行 publisher 時(shí)會彈窗詢問是否允許要點(diǎn)“允許訪問”。如果沒彈窗但 subscriber 收不到消息手動在防火墻入站規(guī)則里加一條 TCP 5555 端口的允許規(guī)則。如果要在兩臺機(jī)器上測試把 subscriber 里的tcp://127.0.0.1:5555改成tcp://發(fā)布端IP:5555同時(shí)確保兩臺機(jī)器在同一網(wǎng)段、發(fā)布端防火墻放行?？鐧C(jī)器測試時(shí)延遲通常在 1ms 以內(nèi)對于工控場景的狀態(tài)廣播完全夠用。4. 避坑排查VS2008 配 ZeroMQ 的五個(gè)血淚經(jīng)驗(yàn)4.1 現(xiàn)象編譯時(shí)報(bào)LNK2005: _sprintf 已經(jīng)在 libcmt.lib 中定義原因libzmq 源碼里有些文件用了sprintf而 VS2008 的 CRT 庫對sprintf有安全警告如果項(xiàng)目里同時(shí)鏈接了libcmt.lib和msvcrt.lib就會報(bào)重復(fù)定義。解決在項(xiàng)目屬性 C/C → 代碼生成 → 運(yùn)行時(shí)庫里統(tǒng)一選“多線程 (/MT)”不要混用。如果第三方庫必須用 /MD那就把 libzmq 也改成 /MD 重新編譯保持所有模塊的 CRT 一致。4.2 現(xiàn)象subscriber 收不到消息但 publisher 顯示發(fā)送成功原因ZeroMQ 的 pub/sub 模式有一個(gè)“慢連接”問題。publisher 綁定端口后subscriber 需要幾毫秒到幾百毫秒的時(shí)間完成 TCP 握手和訂閱信息同步。如果 publisher 在 subscriber 連接完成之前就發(fā)送了消息這些消息會被直接丟棄因?yàn)?pub 端不緩存消息。解決在 publisher 的zmq_bind之后加一個(gè)Sleep(500)等訂閱端連上來再開始發(fā)?；蛘咴?subscriber 端加一個(gè)重試邏輯收到第一條消息后再開始處理。4.3 現(xiàn)象程序退出時(shí)卡在zmq_ctx_destroy原因zmq_ctx_destroy會等待所有 socket 關(guān)閉后才銷毀上下文。如果還有 socket 沒關(guān)或者有線程阻塞在zmq_recv上這個(gè)函數(shù)就會一直等。解決確保每個(gè) socket 都調(diào)用了zmq_close并且zmq_close之后不要再使用這個(gè) socket 指針。如果有接收線程先用zmq_setsockopt設(shè)置ZMQ_LINGER為 0這樣zmq_close會立即返回不會等待未發(fā)送的消息。4.4 現(xiàn)象Release 版在開發(fā)機(jī)上正常拷到工控機(jī)上報(bào)“應(yīng)用程序配置不正確”原因VS2008 編譯的 Release 版默認(rèn)依賴 VC 2008 運(yùn)行庫msvcr90.dll 等工控機(jī)上沒裝這個(gè)運(yùn)行庫。解決在項(xiàng)目屬性里把運(yùn)行時(shí)庫改成“多線程 (/MT)”這樣會把 CRT 靜態(tài)鏈接進(jìn) exe不依賴外部 DLL。注意 libzmq 也要用 /MT 重新編譯否則鏈接時(shí)會沖突。4.5 現(xiàn)象發(fā)送中文消息時(shí) subscriber 收到亂碼原因zmq_send發(fā)送的是原始字節(jié)流不處理字符編碼。如果 publisher 源碼文件是 GBK 編碼而 subscriber 按 UTF-8 解析就會亂碼。解決統(tǒng)一用 UTF-8 編碼保存源碼文件或者在發(fā)送前把字符串轉(zhuǎn)成 UTF-8。VS2008 默認(rèn)保存為 GBK需要在“文件 → 高級保存選項(xiàng)”里改成 UTF-8 無簽名。如果不想改源碼編碼可以在發(fā)送前用MultiByteToWideChar和WideCharToMultiByte做一次轉(zhuǎn)換。5. 進(jìn)階技巧用多線程訂閱和消息分幀提升吞吐5.1 多線程訂閱模型單線程訂閱端在消息量大時(shí)會丟消息因?yàn)閦mq_recv和業(yè)務(wù)處理在同一個(gè)線程里處理慢的時(shí)候接收緩沖區(qū)會滿。常見做法是用兩個(gè)線程一個(gè)線程專門zmq_recv收到消息后放進(jìn)一個(gè)線程安全的隊(duì)列另一個(gè)線程從隊(duì)列取消息做業(yè)務(wù)處理。ZeroMQ 的 socket 不是線程安全的不能兩個(gè)線程同時(shí)操作同一個(gè) socket所以接收線程和處理線程之間要用隊(duì)列傳遞數(shù)據(jù)。// 接收線程只負(fù)責(zé)收消息入隊(duì) DWORD WINAPI recv_thread(LPVOID param) { void* subscriber (void*)param; while (1) { char buffer[256]; int size zmq_recv(subscriber, buffer, sizeof(buffer) - 1, 0); if (size 0) break; buffer[size] \0; // 入隊(duì)操作偽代碼實(shí)際用 CriticalSection 或互斥量保護(hù) EnterCriticalSection(cs); queue.push(std::string(buffer)); LeaveCriticalSection(cs); } return 0; }接收線程里不要做任何耗時(shí)操作zmq_recv之后立刻入隊(duì)然后繼續(xù)下一次接收。隊(duì)列用std::queue加CRITICAL_SECTION保護(hù)VS2008 不支持 C11 的std::mutex只能用 Windows 原生的臨界區(qū)。處理線程從隊(duì)列取數(shù)據(jù)時(shí)如果隊(duì)列為空就Sleep(1)讓出 CPU不要忙等。5.2 消息分幀與多部分消息ZeroMQ 支持多部分消息可以把一個(gè)邏輯消息拆成多個(gè)幀發(fā)送接收端按幀接收。這在發(fā)送結(jié)構(gòu)化數(shù)據(jù)時(shí)很有用比如第一幀是主題第二幀是 JSON 負(fù)載// 發(fā)送端多部分消息 zmq_send(publisher, status, 6, ZMQ_SNDMORE); // 第一幀ZMQ_SNDMORE 表示還有后續(xù)幀 zmq_send(publisher, {\temp\:25}, 11, 0); // 第二幀最后一幀 // 接收端循環(huán)接收直到?jīng)]有更多幀 int more 1; size_t more_size sizeof(more); while (more) { char buffer[256]; int size zmq_recv(subscriber, buffer, sizeof(buffer) - 1, 0); buffer[size] \0; printf(frame: %s\n, buffer); zmq_getsockopt(subscriber, ZMQ_RCVMORE, more, more_size); }ZMQ_SNDMORE標(biāo)志告訴 ZeroMQ 這一幀后面還有幀接收端通過ZMQ_RCVMORE選項(xiàng)判斷是否還有后續(xù)幀。多部分消息的原子性由 ZeroMQ 保證要么全部收到要么全部收不到不會出現(xiàn)只收到一半的情況。這個(gè)特性在發(fā)送配置更新或批量數(shù)據(jù)時(shí)很有用接收端可以等所有幀到齊后再統(tǒng)一處理。5.3 用zmq_poll做超時(shí)接收阻塞式zmq_recv在沒有消息時(shí)會一直等程序無法響應(yīng)退出信號。用zmq_poll可以設(shè)置超時(shí)時(shí)間超時(shí)后返回 0程序可以檢查退出標(biāo)志zmq_pollitem_t items[] { { subscriber, 0, ZMQ_POLLIN, 0 } }; int rc zmq_poll(items, 1, 100); // 超時(shí) 100ms if (rc 0 (items[0].revents ZMQ_POLLIN)) { // 有消息可收 char buffer[256]; int size zmq_recv(subscriber, buffer, sizeof(buffer) - 1, 0); // 處理消息 } else if (rc 0) { // 超時(shí)檢查退出標(biāo)志 if (g_exit) break; }zmq_pollitem_t的第一個(gè)字段是 socket 指針第二個(gè)是文件描述符Windows 下填 0第三個(gè)是監(jiān)聽的事件類型第四個(gè)是返回的事件類型。zmq_poll的第三個(gè)參數(shù)是超時(shí)毫秒數(shù)-1 表示無限等待0 表示立即返回。這個(gè)函數(shù)在需要同時(shí)監(jiān)聽多個(gè) socket 時(shí)特別有用可以把訂閱 socket 和命令 socket 放在同一個(gè) poll 數(shù)組里。從那以后我每次在 VS2008 里配 ZeroMQ都會先把ZMQ_STATIC和/MT這兩個(gè)編譯選項(xiàng)確認(rèn)一遍再寫一個(gè)zmq_version的測試工程驗(yàn)證鏈接最后才動 pub/sub 的代碼。這個(gè)順序能省掉八成以上的鏈接錯(cuò)誤和運(yùn)行時(shí)崩潰。希望幫到你。本文還有配套的精品資源點(diǎn)擊獲取