# ZMQ 指南
**作者: Pieter Hintjens <ph@imatix.com>, CEO iMatix Corporation.**
**翻譯: 張吉 <jizhang@anjuke.com>, 安居客集團 好租網工程師**
With thanks to Bill Desmarais, Brian Dorsey, CAF, Daniel Lin, Eric Desgranges, Gonzalo Diethelm, Guido Goldstein, Hunter Ford, Kamil Shakirov, Martin Sustrik, Mike Castleman, Naveen Chawla, Nicola Peduzzi, Oliver Smith, Olivier Chamoux, Peter Alexander, Pierre Rouleau, Randy Dryburgh, John Unwin, Alex Thomas, Mihail Minkov, Jeremy Avnet, Michael Compton, Kamil Kisiel, Mark Kharitonov, Guillaume Aubert, Ian Barber, Mike Sheridan, Faruk Akgul, Oleg Sidorov, Lev Givon, Allister MacLeod, Alexander?D'Archangel, Andreas Hoelzlwimmer, Han Holl, Robert G. Jakabosky, Felipe Cruz, Marcus McCurdy, Mikhail Kulemin, Dr. Gerg? érdi, Pavel Zhukov, Alexander Else, Giovanni Ruggiero, Rick "Technoweenie", Daniel Lundin, Dave Hoover, Simon Jefford, Benjamin Peterson, Justin Case, Devon Weller, Richard Smith, Alexander Morland, Wadim Grasza, Michael Jakl, and Zed Shaw for their contributions, and to Stathis Sideris for [Ditaa](http://www.ditaa.org).
Please use the [issue tracker](https://github.com/imatix/zguide/issues) for all comments and errata. This version covers the latest stable release of 0MQ and was published on Mon 10 October, 2011.
The Guide is mainly [in C](http://zguide.zeromq.org/page:all), but also in [PHP](http://zguide.zeromq.org/php:all) and [Lua](http://zguide.zeromq.org/lua:all).
---
This work is licensed under a [Creative Commons Attribution-ShareAlike 3.0 License](http://creativecommons.org/licenses/by-sa/3.0/).
## 第一章 ZeroMQ基礎
### 拯救世界
如何解釋ZMQ？有些人會先說一堆ZMQ的好：它是一套用于快速構建的套接字組件；它的信箱系統有超強的路由能力；它太快了！而有些人則喜歡分享他們被ZMQ點悟的時刻，那些被靈感擊中的瞬間：所有的事情突然變得簡單明了，讓人大開眼界。另一些人則會拿ZMQ同其他產品做個比較：它更小，更簡單，但卻讓人覺得如此熟悉。對于我個人而言，我則更傾向于和別人分享ZMQ的誕生史，相信會和各位讀者有所共鳴。
編程是一門科學，但往往會喬裝成一門藝術。我們從不去了解軟件最底層的機理，或者說根本沒有人在乎這些。軟件并不只是算法、數據結構、編程語言、或者抽象云云，這些不過是一些工具而已，被我們創造、使用、最后拋棄。軟件真正的本質，其實是人的本質。
舉例來說，當我們遇到一個高度復雜的問題時，我們會群策群力，分工合作，將問題拆分為若干個部分，一起解決。這里就體現了編程的科學：創建一組小型的構建模塊，讓人們易于理解和使用，那么大家就會一起用它來解決問題。
我們生活在一個普遍聯系的世界里，需要現代的編程軟件為我們做指引。所以，未來我們所需要的用于處理大規模計算的構建模塊，必須是普遍聯系的，而且能夠并行運作。那時，程序代碼不能再只關注自己，它們需要互相交流，變得足夠健談。程序代碼需要像人腦一樣，數以兆計的神經元高速地傳輸信號，在一個沒有中央控制的環境下，沒有單點故障的環境下，解決問題。這一點其實并不意外，因為就當今的網絡來講，每個節點其實就像是連接了一個人腦一樣。
如果你曾和線程、協議、或網絡打過交道，你會覺得我上面的話像是天方夜譚。因為在實際應用過程中，只是連接幾個程序或網絡就已經非常困難和麻煩了。數以兆計的節點？那真是無法想象的。現今只有資金雄厚的企業才能負擔得起這種軟件和服務。
當今世界的網絡結構已經遠遠超越了我們自身的駕馭能力。十九世紀八十年代的軟件危機，弗萊德?布魯克斯曾說過，這個世上[沒有銀彈](http://en.wikipedia.org/wiki/No_Silver_Bullet
)。后來，免費和開源解決了這次軟件危機，讓我們能夠高效地分享知識。如今，我們又面臨一次新的軟件危機，只不過我們談論得不多。只有那些大型的、富足的企業才有財力建立高度聯系的應用程序。那里有云的存在，但它是私有的。我們的數據和知識正在從我們的個人電腦中消失，流入云端，無法獲得或與其競爭。是誰坐擁我們的社交網絡？這真像一次巨型主機的革命。
我們暫且不談其中的政治因素，光那些就可以另外出本書了。目前的現狀是，雖然互聯網能夠讓千萬個程序相連，但我們之中的大多數卻無法做到這些。這樣一來，那些真正有趣的大型問題（如健康、教育、經濟、交通等領域），仍然無法解決。我們沒有能力將代碼連接起來，也就不能像大腦中的神經元一樣處理那些大規模的問題。
已經有人嘗試用各種方法來連接應用程序，如數以千計的IETF規范，每種規范解決一個特定問題。對于開發人員來說，HTTP協議是比較簡單和易用的，但這也往往讓問題變得更糟，因為它鼓勵人們形成一種重服務端、輕客戶端的思想。
所以迄今為止人們還在使用原始的TCP/UDP協議、私有協議、HTTP協議、網絡套接字等形式連接應用程序。這種做法依舊讓人痛苦，速度慢又不易擴展，需要集中化管理。而分布式的P2P協議又僅僅適用于娛樂，而非真正的應用。有誰會使用Skype或者Bittorrent來交換數據呢？
這就讓我們回歸到編程科學的問題上來。想要拯救這個世界，我們需要做兩件事情：一，如何在任何地點連接任何兩個應用程序；二、將這個解決方案用最為簡單的方式包裝起來，供程序員使用。
也許這聽起來太簡單了，但事實確實如此。
### ZMQ簡介
ZMQ（?MQ、ZeroMQ, 0MQ）看起來像是一套嵌入式的網絡鏈接庫，但工作起來更像是一個并發式的框架。它提供的套接字可以在多種協議中傳輸消息，如線程間、進程間、TCP、廣播等。你可以使用套接字構建多對多的連接模式，如扇出、發布-訂閱、任務分發、請求-應答等。ZMQ的快速足以勝任集群應用產品。它的異步I/O機制讓你能夠構建多核應用程序，完成異步消息處理任務。ZMQ有著多語言支持，并能在幾乎所有的操作系統上運行。ZMQ是[iMatix][]公司的產品，以LGPL開源協議發布。
### 需要具備的知識
* 使用最新的ZMQ穩定版本；
* 使用Linux系統或其他相似的操作系統；
* 能夠閱讀C語言代碼，這是本指南示例程序的默認語言；
* 當我們書寫諸如PUSH或SUBSCRIBE等常量時，你能夠找到相應語言的實現，如ZMQ_PUSH、ZMQ_SUBSCRIBE。
### 獲取示例
本指南的所有示例都存放于[github倉庫](https://github.com/imatix/zguide)中，最簡單的獲取方式是運行以下代碼：
```
git clone git://github.com/imatix/zguide.git
```
瀏覽examples目錄，你可以看到多種語言的實現。如果其中缺少了某種你正在使用的語言，我們很希望你可以[提交一份補充](http://zguide.zeromq.org/main:translate)。這也是本指南實用的原因，要感謝所有做出過貢獻的人。
所有的示例代碼都以MIT/X11協議發布，若在源代碼中有其他限定的除外。
### 提問-回答
讓我們從簡單的代碼開始，一段傳統的Hello World程序。我們會創建一個客戶端和一個服務端，客戶端發送Hello給服務端，服務端返回World。下文是C語言編寫的服務端，它在5555端口打開一個ZMQ套接字，等待請求，收到后應答World。
**hwserver.c: Hello World server**
```c
//
// Hello World 服務端
// 綁定一個REP套接字至tcp://*:5555
// 從客戶端接收Hello，并應答World
//
#include <zmq.h>
#include <stdio.h>
#include <unistd.h>
#include <string.h>
int main (void)
{
void *context = zmq_init (1);
// 與客戶端通信的套接字
void *responder = zmq_socket (context, ZMQ_REP);
zmq_bind (responder, "tcp://*:5555");
while (1) {
// 等待客戶端請求
zmq_msg_t request;
zmq_msg_init (&request);
zmq_recv (responder, &request, 0);
printf ("收到 Hello\n");
zmq_msg_close (&request);
// 做些“處理”
sleep (1);
// 返回應答
zmq_msg_t reply;
zmq_msg_init_size (&reply, 5);
memcpy (zmq_msg_data (&reply), "World", 5);
zmq_send (responder, &reply, 0);
zmq_msg_close (&reply);
}
// 程序不會運行到這里，以下只是演示我們應該如何結束
zmq_close (responder);
zmq_term (context);
return 0;
}
```
使用REQ-REP套接字發送和接受消息是需要遵循一定規律的。客戶端首先使用zmq_send()發送消息，再用zmq_recv()接收，如此循環。如果打亂了這個順序（如連續發送兩次）則會報錯。類似地，服務端必須先進行接收，后進行發送。
ZMQ使用C語言作為它參考手冊的語言，本指南也以它作為示例程序的語言。如果你正在閱讀本指南的在線版本，你可以看到示例代碼的下方有其他語言的實現。如以下是C++語言：
**hwserver.cpp: Hello World server**
```cpp
//
// Hello World 服務端 C++語言版
// 綁定一個REP套接字至tcp://*:5555
// 從客戶端接收Hello，并應答World
//
#include <zmq.hpp>
#include <string>
#include <iostream>
#include <unistd.h>
int main () {
// 準備上下文和套接字
zmq::context_t context (1);
zmq::socket_t socket (context, ZMQ_REP);
socket.bind ("tcp://*:5555");
while (true) {
zmq::message_t request;
// 等待客戶端請求
socket.recv (&request);
std::cout << "收到 Hello" << std::endl;
// 做一些“處理”
sleep (1);
// 應答World
zmq::message_t reply (5);
memcpy ((void *) reply.data (), "World", 5);
socket.send (reply);
}
return 0;
}
```
可以看到C語言和C++語言的API代碼差不多，而在PHP這樣的語言中，代碼就會更為簡潔：
**hwserver.php: Hello World server**
```php
<?php
/**
* Hello World 服務端
* 綁定REP套接字至 tcp://*:5555
* 從客戶端接收Hello，并應答World
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext(1);
// 與客戶端通信的套接字
$responder = new ZMQSocket($context, ZMQ::SOCKET_REP);
$responder->bind("tcp://*:5555");
while(true) {
// 等待客戶端請求
$request = $responder->recv();
printf ("Received request: [%s]\n", $request);
// 做一些“處理”
sleep (1);
// 應答World
$responder->send("World");
}
```
下面是客戶端的代碼：
**hwclient: Hello World client in C**
```c
//
// Hello World 客戶端
// 連接REQ套接字至 tcp://localhost:5555
// 發送Hello給服務端，并接收World
//
#include <zmq.h>
#include <string.h>
#include <stdio.h>
#include <unistd.h>
int main (void)
{
void *context = zmq_init (1);
// 連接至服務端的套接字
printf ("正在連接至hello world服務端...\n");
void *requester = zmq_socket (context, ZMQ_REQ);
zmq_connect (requester, "tcp://localhost:5555");
int request_nbr;
for (request_nbr = 0; request_nbr != 10; request_nbr++) {
zmq_msg_t request;
zmq_msg_init_size (&request, 5);
memcpy (zmq_msg_data (&request), "Hello", 5);
printf ("正在發送 Hello %d...\n", request_nbr);
zmq_send (requester, &request, 0);
zmq_msg_close (&request);
zmq_msg_t reply;
zmq_msg_init (&reply);
zmq_recv (requester, &reply, 0);
printf ("接收到 World %d\n", request_nbr);
zmq_msg_close (&reply);
}
zmq_close (requester);
zmq_term (context);
return 0;
}
```
這看起來是否太簡單了？ZMQ就是這樣一個東西，你往里加點兒料就能制作出一枚無窮能量的原子彈，用它來拯救世界吧！
理論上你可以連接千萬個客戶端到這個服務端上，同時連接都沒問題，程序仍會運作得很好。你可以嘗試一下先打開客戶端，再打開服務端，可以看到程序仍然會正常工作，想想這意味著什么。
讓我簡單介紹一下這兩段程序到底做了什么。首先，他們創建了一個ZMQ上下文，然后是一個套接字。不要被這些陌生的名詞嚇到，后面我們都會講到。服務端將REP套接字綁定到5555端口上，并開始等待請求，發出應答，如此循環。客戶端則是發送請求并等待服務端的應答。
這些代碼背后其實發生了很多很多事情，但是程序員完全不必理會這些，只要知道這些代碼短小精悍，極少出錯，耐高壓。這種通信模式我們稱之為請求-應答模式，是ZMQ最直接的一種應用。你可以拿它和RPC及經典的C/S模型做類比。
### 關于字符串
ZMQ不會關心發送消息的內容，只要知道它所包含的字節數。所以，程序員需要做一些工作，保證對方節點能夠正確讀取這些消息。如何將一個對象或復雜數據類型轉換成ZMQ可以發送的消息，這有類似Protocol Buffers的序列化軟件可以做到。但對于字符串，你也是需要有所注意的。
在C語言中，字符串都以一個空字符結尾，你可以像這樣發送一個完整的字符串：
```c
zmq_msg_init_data (&request, "Hello", 6, NULL, NULL);
```
但是，如果你用其他語言發送這個字符串，很可能不會包含這個空字節，如你使用Python發送：
```python
socket.send ("Hello")
```
實際發送的消息是：
如果你從C語言中讀取該消息，你會讀到一個類似于字符串的內容，甚至它可能就是一個字符串（第六位在內存中正好是一個空字符），但是這并不合適。這樣一來，客戶端和服務端對字符串的定義就不統一了，你會得到一些奇怪的結果。
當你用C語言從ZMQ中獲取字符串，你不能夠相信該字符串有一個正確的結尾。因此，當你在接受字符串時，應該建立多一個字節的緩沖區，將字符串放進去，并添加結尾。
所以，讓我們做如下假設：**ZMQ的字符串是有長度的，且傳送時不加結束符**。在最簡單的情況下，ZMQ字符串和ZMQ消息中的一幀是等價的，就如上圖所展現的，由一個長度屬性和一串字節表示。
下面這個功能函數會幫助我們在C語言中正確的接受字符串消息：
```c
// 從ZMQ套接字中接收字符串，并轉換為C語言的字符串
static char *
s_recv (void *socket) {
zmq_msg_t message;
zmq_msg_init (&message);
zmq_recv (socket, &message, 0);
int size = zmq_msg_size (&message);
char *string = malloc (size + 1);
memcpy (string, zmq_msg_data (&message), size);
zmq_msg_close (&message);
string [size] = 0;
return (string);
}
```
這段代碼我們會在日后的示例中使用，我們可以順手寫一個s_send()方法，并打包成一個.h文件供我們使用。
這就誕生了zhelpers.h，一個供C語言使用的ZMQ功能函數庫。它的源代碼比較長，而且只對C語言程序員有用，你可以在閑暇時[看一看](https://github.com/imatix/zguide/blob/master/examples/C/zhelpers.h)。
### 獲取版本號
ZMQ目前有多個版本，而且仍在持續更新。如果你遇到了問題，也許這在下一個版本中已經解決了。想知道目前的ZMQ版本，你可以在程序中運行如下：
**version: ?MQ version reporting in C**
```c
//
// 返回當前ZMQ的版本號
//
#include "zhelpers.h"
int main (void)
{
int major, minor, patch;
zmq_version (&major, &minor, &patch);
printf ("當前ZMQ版本號為 %d.%d.%d\n", major, minor, patch);
return EXIT_SUCCESS;
}
```
### 讓消息流動起來
第二種經典的消息模式是單向數據分發：服務端將更新事件發送給一組客戶端。讓我們看一個天氣信息發布的例子，包括郵編、溫度、相對濕度。我們生成這些隨機信息，用來模擬氣象站所做的那樣。
下面是服務端的代碼，使用5556端口：
**wuserver: Weather update server in C**
```c
//
// 氣象信息更新服務
// 綁定PUB套接字至tcp://*:5556端點
// 發布隨機氣象信息
//
#include "zhelpers.h"
int main (void)
{
// 準備上下文和PUB套接字
void *context = zmq_init (1);
void *publisher = zmq_socket (context, ZMQ_PUB);
zmq_bind (publisher, "tcp://*:5556");
zmq_bind (publisher, "ipc://weather.ipc");
// 初始化隨機數生成器
srandom ((unsigned) time (NULL));
while (1) {
// 生成數據
int zipcode, temperature, relhumidity;
zipcode = randof (100000);
temperature = randof (215) - 80;
relhumidity = randof (50) + 10;
// 向所有訂閱者發送消息
char update [20];
sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);
s_send (publisher, update);
}
zmq_close (publisher);
zmq_term (context);
return 0;
}
```
這項更新服務沒有開始、沒有結束，就像永不消失的電波一樣。
下面是客戶端程序，它會接受發布者的消息，只處理特定郵編標注的信息，如紐約的郵編是10001:
**wuclient: Weather update client in C**
```c
//
// 氣象信息客戶端
// 連接SUB套接字至tcp://*:5556端點
// 收集指定郵編的氣象信息，并計算平均溫度
//
#include "zhelpers.h"
int main (int argc, char *argv [])
{
void *context = zmq_init (1);
// 創建連接至服務端的套接字
printf ("正在收集氣象信息...\n");
void *subscriber = zmq_socket (context, ZMQ_SUB);
zmq_connect (subscriber, "tcp://localhost:5556");
// 設置訂閱信息，默認為紐約，郵編10001
char *filter = (argc > 1)? argv [1]: "10001 ";
zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE, filter, strlen (filter));
// 處理100條更新信息
int update_nbr;
long total_temp = 0;
for (update_nbr = 0; update_nbr < 100; update_nbr++) {
char *string = s_recv (subscriber);
int zipcode, temperature, relhumidity;
sscanf (string, "%d %d %d",
&zipcode, &temperature, &relhumidity);
total_temp += temperature;
free (string);
}
printf ("地區郵編 '%s' 的平均溫度為 %dF\n",
filter, (int) (total_temp / update_nbr));
zmq_close (subscriber);
zmq_term (context);
return 0;
}
```
需要注意的是，在使用SUB套接字時，必須使用zmq_setsockopt()方法來設置訂閱的內容。如果你不設置訂閱內容，那將什么消息都收不到，新手很容易犯這個錯誤。訂閱信息可以是任何字符串，可以設置多次。只要消息滿足其中一條訂閱信息，SUB套接字就會收到。訂閱者可以選擇不接收某類消息，也是通過zmq_setsockopt()方法實現的。
PUB-SUB套接字組合是異步的。客戶端在一個循環體中使用zmq_recv()接收消息，如果向SUB套接字發送消息則會報錯；類似地，服務端可以不斷地使用zmq_send()發送消息，但不能在PUB套接字上使用zmq_recv()。
關于PUB-SUB套接字，還有一點需要注意：你無法得知SUB是何時開始接收消息的。就算你先打開了SUB套接字，后打開PUB發送消息，這時SUB還是會丟失一些消息的，因為建立連接是需要一些時間的。很少，但并不是零。
這種“慢連接”的癥狀一開始會讓很多人困惑，所以這里我要詳細解釋一下。還記得ZMQ是在后臺進行異步的I/O傳輸的，如果你有兩個節點用以下順序相連：
* 訂閱者連接至端點接收消息并計數；
* 發布者綁定至端點并立刻發送1000條消息。
運行的結果很可能是訂閱者一條消息都收不到。這時你可能會傻眼，忙于檢查有沒有設置訂閱信息，并重新嘗試，但結果還是一樣。
我們知道在建立TCP連接時需要進行三次握手，會耗費幾毫秒的時間，而當節點數增加時這個數字也會上升。在這么短的時間里，ZMQ就可以發送很多很多消息了。舉例來說，如果建立連接需要耗時5毫秒，而ZMQ只需要1毫秒就可以發送完這1000條消息。
第二章中我會解釋如何使發布者和訂閱者同步，只有當訂閱者準備好時發布者才會開始發送消息。有一種簡單的方法來同步PUB和SUB，就是讓PUB延遲一段時間再發送消息。現實編程中我不建議使用這種方式，因為它太脆弱了，而且不好控制。不過這里我們先暫且使用sleep的方式來解決，等到第二章的時候再講述正確的處理方式。
另一種同步的方式則是認為發布者的消息流是無窮無盡的，因此丟失了前面一部分信息也沒有關系。我們的氣象信息客戶端就是這么做的。
示例中的氣象信息客戶端會收集指定郵編的一千條信息，其間大約有1000萬條信息被發布。你可以先打開客戶端，再打開服務端，工作一段時間后重啟服務端，這時客戶端仍會正常工作。當客戶端收集完所需信息后，會計算并輸出平均溫度。
關于發布-訂閱模式的幾點說明：
* 訂閱者可以連接多個發布者，輪流接收消息；
* 如果發布者沒有訂閱者與之相連，那它發送的消息將直接被丟棄；
* 如果你使用TCP協議，那當訂閱者處理速度過慢時，消息會在發布者處堆積。以后我們會討論如何使用閾值（HWM）來保護發布者。
* 在目前版本的ZMQ中，消息的過濾是在訂閱者處進行的。也就是說，發布者會向訂閱者發送所有的消息，訂閱者會將未訂閱的消息丟棄。
我在自己的四核計算機上嘗試發布1000萬條消息，速度很快，但沒什么特別的：
```
ph@ws200901:~/work/git/0MQGuide/examples/c$ time wuclient
Collecting updates from weather server...
Average temperature for zipcode '10001 ' was 18F
real 0m5.939s
user 0m1.590s
sys 0m2.290s
```
### 分布式處理
下面一個示例程序中，我們將使用ZMQ進行超級計算，也就是并行處理模型：
* 任務分發器會生成大量可以并行計算的任務；
* 有一組worker會處理這些任務；
* 結果收集器會在末端接收所有worker的處理結果，進行匯總。
現實中，worker可能散落在不同的計算機中，利用GPU（圖像處理單元）進行復雜計算。下面是任務分發器的代碼，它會生成100個任務，任務內容是讓收到的worker延遲若干毫秒。
**taskvent: Parallel task ventilator in C**
```c
//
// 任務分發器
// 綁定PUSH套接字至tcp://localhost:5557端點
// 發送一組任務給已建立連接的worker
//
#include "zhelpers.h"
int main (void)
{
void *context = zmq_init (1);
// 用于發送消息的套接字
void *sender = zmq_socket (context, ZMQ_PUSH);
zmq_bind (sender, "tcp://*:5557");
// 用于發送開始信號的套接字
void *sink = zmq_socket (context, ZMQ_PUSH);
zmq_connect (sink, "tcp://localhost:5558");
printf ("準備好worker后按任意鍵開始: ");
getchar ();
printf ("正在向worker分配任務...\n");
// 發送開始信號
s_send (sink, "0");
// 初始化隨機數生成器
srandom ((unsigned) time (NULL));
// 發送100個任務
int task_nbr;
int total_msec = 0; // 預計執行時間（毫秒）
for (task_nbr = 0; task_nbr < 100; task_nbr++) {
int workload;
// 隨機產生1-100毫秒的工作量
workload = randof (100) + 1;
total_msec += workload;
char string [10];
sprintf (string, "%d", workload);
s_send (sender, string);
}
printf ("預計執行時間: %d 毫秒\n", total_msec);
sleep (1); // 延遲一段時間，讓任務分發完成
zmq_close (sink);
zmq_close (sender);
zmq_term (context);
return 0;
}
```
下面是worker的代碼，它接受信息并延遲指定的毫秒數，并發送執行完畢的信號：
**taskwork: Parallel task worker in C**
```c
//
// 任務執行器
// 連接PULL套接字至tcp://localhost:5557端點
// 從任務分發器處獲取任務
// 連接PUSH套接字至tcp://localhost:5558端點
// 向結果采集器發送結果
//
#include "zhelpers.h"
int main (void)
{
void *context = zmq_init (1);
// 獲取任務的套接字
void *receiver = zmq_socket (context, ZMQ_PULL);
zmq_connect (receiver, "tcp://localhost:5557");
// 發送結果的套接字
void *sender = zmq_socket (context, ZMQ_PUSH);
zmq_connect (sender, "tcp://localhost:5558");
// 循環處理任務
while (1) {
char *string = s_recv (receiver);
// 輸出處理進度
fflush (stdout);
printf ("%s.", string);
// 開始處理
s_sleep (atoi (string));
free (string);
// 發送結果
s_send (sender, "");
}
zmq_close (receiver);
zmq_close (sender);
zmq_term (context);
return 0;
}
```
下面是結果收集器的代碼。它會收集100個處理結果，并計算總的執行時間，讓我們由此判別任務是否是并行計算的。
**tasksink: Parallel task sink in C**
```c
//
// 任務收集器
// 綁定PULL套接字至tcp://localhost:5558端點
// 從worker處收集處理結果
//
#include "zhelpers.h"
int main (void)
{
// 準備上下文和套接字
void *context = zmq_init (1);
void *receiver = zmq_socket (context, ZMQ_PULL);
zmq_bind (receiver, "tcp://*:5558");
// 等待開始信號
char *string = s_recv (receiver);
free (string);
// 開始計時
int64_t start_time = s_clock ();
// 確定100個任務均已處理
int task_nbr;
for (task_nbr = 0; task_nbr < 100; task_nbr++) {
char *string = s_recv (receiver);
free (string);
if ((task_nbr / 10) * 10 == task_nbr)
printf (":");
else
printf (".");
fflush (stdout);
}
// 計算并輸出總執行時間
printf ("執行時間: %d 毫秒\n",
(int) (s_clock () - start_time));
zmq_close (receiver);
zmq_term (context);
return 0;
}
```
一組任務的平均執行時間在5秒左右，以下是分別開始1個、2個、4個worker時的執行結果：
```
# 1 worker
Total elapsed time: 5034 msec
# 2 workers
Total elapsed time: 2421 msec
# 4 workers
Total elapsed time: 1018 msec
```
關于這段代碼的幾個細節：
* worker上游和任務分發器相連，下游和結果收集器相連，這就意味著你可以開啟任意多個worker。但若worker是綁定至端點的，而非連接至端點，那我們就需要準備更多的端點，并配置任務分發器和結果收集器。所以說，任務分發器和結果收集器是這個網絡結構中較為穩定的部分，因此應該由它們綁定至端點，而非worker，因為它們較為動態。
* 我們需要做一些同步的工作，等待worker全部啟動之后再分發任務。這點在ZMQ中很重要，且不易解決。連接套接字的動作會耗費一定的時間，因此當第一個worker連接成功時，它會一下收到很多任務。所以說，如果我們不進行同步，那這些任務根本就不會被并行地執行。你可以自己試驗一下。
* 任務分發器使用PUSH套接字向worker均勻地分發任務（假設所有的worker都已經連接上了），這種機制稱為_負載均衡_，以后我們會見得更多。
* 結果收集器的PULL套接字會均勻地從worker處收集消息，這種機制稱為_公平隊列_：
管道模式也會出現慢連接的情況，讓人誤以為PUSH套接字沒有進行負載均衡。如果你的程序中某個worker接收到了更多的請求，那是因為它的PULL套接字連接得比較快，從而在別的worker連接之前獲取了額外的消息。
### 使用ZMQ編程
看著這些示例程序后，你一定迫不及待想要用ZMQ進行編程了。不過在開始之前，我還有幾條建議想給到你，這樣可以省去未來的一些麻煩：
* 學習ZMQ要循序漸進，雖然它只是一套API，但卻提供了無盡的可能。一步一步學習它提供的功能，并完全掌握。
* 編寫漂亮的代碼。丑陋的代碼會隱藏問題，讓想要幫助你的人無從下手。比如，你會習慣于使用無意義的變量名，但讀你代碼的人并不知道。應使用有意義的變量名稱，而不是隨意起一個。代碼的縮進要統一，布局清晰。漂亮的代碼可以讓你的世界變得更美好。
* 邊寫邊測試，當代碼出現問題，你就可以快速定位到某些行。這一點在編寫ZMQ應用程序時尤為重要，因為很多時候你無法第一次就編寫出正確的代碼。
* 當你發現自己編寫的代碼無法正常工作時，你可以將其拆分成一些代碼片段，看看哪段沒有正確地執行。ZMQ可以讓你構建非常模塊化的代碼，所以應該好好利用這一點。
* 需要時應使用抽象的方法來編寫程序（類、成員函數等等），不要隨意拷貝代碼，因為拷貝代碼的同時也是在拷貝錯誤。
我們看看下面這段代碼，是某位同仁讓我幫忙修改的：
```c
// 注意：不要使用這段代碼！
static char *topic_str = "msg.x|";
void* pub_worker(void* arg){
void *ctx = arg;
assert(ctx);
void *qskt = zmq_socket(ctx, ZMQ_REP);
assert(qskt);
int rc = zmq_connect(qskt, "inproc://querys");
assert(rc == 0);
void *pubskt = zmq_socket(ctx, ZMQ_PUB);
assert(pubskt);
rc = zmq_bind(pubskt, "inproc://publish");
assert(rc == 0);
uint8_t cmd;
uint32_t nb;
zmq_msg_t topic_msg, cmd_msg, nb_msg, resp_msg;
zmq_msg_init_data(&topic_msg, topic_str, strlen(topic_str) , NULL, NULL);
fprintf(stdout,"WORKER: ready to recieve messages\n");
// 注意：不要使用這段代碼，它不能工作！
// e.g. topic_msg will be invalid the second time through
while (1){
zmq_send(pubskt, &topic_msg, ZMQ_SNDMORE);
zmq_msg_init(&cmd_msg);
zmq_recv(qskt, &cmd_msg, 0);
memcpy(&cmd, zmq_msg_data(&cmd_msg), sizeof(uint8_t));
zmq_send(pubskt, &cmd_msg, ZMQ_SNDMORE);
zmq_msg_close(&cmd_msg);
fprintf(stdout, "recieved cmd %u\n", cmd);
zmq_msg_init(&nb_msg);
zmq_recv(qskt, &nb_msg, 0);
memcpy(&nb, zmq_msg_data(&nb_msg), sizeof(uint32_t));
zmq_send(pubskt, &nb_msg, 0);
zmq_msg_close(&nb_msg);
fprintf(stdout, "recieved nb %u\n", nb);
zmq_msg_init_size(&resp_msg, sizeof(uint8_t));
memset(zmq_msg_data(&resp_msg), 0, sizeof(uint8_t));
zmq_send(qskt, &resp_msg, 0);
zmq_msg_close(&resp_msg);
}
return NULL;
}
```
下面是我為他重寫的代碼，順便修復了一些BUG：
```c
static void *
worker_thread (void *arg) {
void *context = arg;
void *worker = zmq_socket (context, ZMQ_REP);
assert (worker);
int rc;
rc = zmq_connect (worker, "ipc://worker");
assert (rc == 0);
void *broadcast = zmq_socket (context, ZMQ_PUB);
assert (broadcast);
rc = zmq_bind (broadcast, "ipc://publish");
assert (rc == 0);
while (1) {
char *part1 = s_recv (worker);
char *part2 = s_recv (worker);
printf ("Worker got [%s][%s]\n", part1, part2);
s_sendmore (broadcast, "msg");
s_sendmore (broadcast, part1);
s_send (broadcast, part2);
free (part1);
free (part2);
s_send (worker, "OK");
}
return NULL;
}
```
上段程序的最后，它將套接字在兩個線程之間傳遞，這會導致莫名其妙的問題。這種行為在ZMQ 2.1中雖然是合法的，但是不提倡使用。
### ZMQ 2.1版
歷史告訴我們，ZMQ 2.0是一個低延遲的分布式消息系統，它從眾多同類軟件中脫穎而出，擺脫了各種奢華的名目，向世界宣告“無極限”的口號。這是我們一直在使用的穩定發行版。
時過境遷，2010年流行的東西在2011年就不一定了。當ZMQ的開發者和社區開發者在激烈地討論ZMQ的種種問題時，ZMQ 2.1橫空出世了，成為新的穩定發行版。
本指南主要針對ZMQ 2.1進行描述，因此對于從ZMQ 2.0遷移過來的開發者來說有一些需要注意的地方：
* 在2.0中，調用zmq_close()和zmq_term()時會丟棄所有尚未發送的消息，所以在發送完消息后不能直接關閉程序，2.0的示例中往往使用sleep(1)來規避這個問題。但是在2.1中就不需要這樣做了，程序會等待消息全部發送完畢后再退出。
* 相反地，2.0中可以在尚有套接字打開的情況下調用zmq_term()，這在2.1中會變得不安全，會造成程序的阻塞。所以，在2.1程序中我們_會先關閉所有的套接字_，然后才退出程序。如果套接字中有尚未發送的消息，程序就會一直處于等待狀態，_除非手工設置了套接字的LINGER選項_（如設置為零），那么套接字會在相應的時間后關閉。
```c
int zero = 0;
zmq_setsockopt (mysocket, ZMQ_LINGER, &zero, sizeof (zero));
```
* 2.0中，zmq_poll()函數沒有定時功能，它會在滿足條件時立刻返回，我們需要在循環體中檢查還有多少剩余。但在2.1中，zmq_poll()會在指定時間后返回，因此可以作為定時器使用。
* 2.0中，ZMQ會忽略系統的中斷消息，這就意味著對libzmq的調用是不會收到EINTR消息的，這樣就無法對SIGINT（Ctrl-C）等消息進行處理了。在2.1中，這個問題得以解決，像類似zmq_recv()的方法都會接收并返回系統的EINTR消息。
### 正確地使用上下文
ZMQ應用程序的一開始總是會先創建一個上下文，并用它來創建套接字。在C語言中，創建上下文的函數是zmq_init()。一個進程中只應該創建一個上下文。從技術的角度來說，上下文是一個容器，包含了該進程下所有的套接字，并為inproc協議提供實現，用以高速連接進程內不同的線程。如果一個進程中創建了兩個上下文，那就相當于啟動了兩個ZMQ實例。如果這正是你需要的，那沒有問題，但一般情況下：
**在一個進程中使用zmq_init()函數創建一個上下文，并在結束時使用zmq_term()函數關閉它**
如果你使用了fork()系統調用，那每個進程需要自己的上下文對象。如果在調用fork()之前調用了zmq_init()函數，那每個子進程都會有自己的上下文對象。通常情況下，你會需要在子進程中做些有趣的事，而讓父進程來管理它們。
### 正確地退出和清理
程序員的一個良好習慣是：總是在結束時進行清理工作。當你使用像Python那樣的語言編寫ZMQ應用程序時，系統會自動幫你完成清理。但如果使用的是C語言，那就需要小心地處理了，否則可能發生內存泄露、應用程序不穩定等問題。
內存泄露只是問題之一，其實ZMQ是很在意程序的退出方式的。個中原因比較復雜，但簡單的來說，如果仍有套接字處于打開狀態，調用zmq_term()時會導致程序掛起；就算關閉了所有的套接字，如果仍有消息處于待發送狀態，zmq_term()也會造成程序的等待。只有當套接字的LINGER選項設為0時才能避免。
我們需要關注的ZMQ對象包括：消息、套接字、上下文。好在內容并不多，至少在一般的應用程序中是這樣：
* 處理完消息后，記得用zmq_msg_close()函數關閉消息；
* 如果你同時打開或關閉了很多套接字，那可能需要重新規劃一下程序的結構了；
* 退出程序時，應該先關閉所有的套接字，最后調用zmq_term()函數，銷毀上下文對象。
如果要用ZMQ進行多線程的編程，需要考慮的問題就更多了。我們會在下一章中詳述多線程編程，但如果你耐不住性子想要嘗試一下，以下是在退出時的一些建議：
* 不要在多個線程中使用同一個套接字。不要去想為什么，反正別這么干就是了。
* 關閉所有的套接字，并在主程序中關閉上下文對象。
* 如果仍有處于阻塞狀態的recv或poll調用，應該在主程序中捕捉這些錯誤，并在相應的線程中關閉套接字。不要重復關閉上下文，zmq_term()函數會等待所有的套接字安全地關閉后才結束。
看吧，過程是復雜的，所以不同語言的API實現者可能會將這些步驟封裝起來，讓結束程序變得不那么復雜。
### 我們為什么需要ZMQ
現在我們已經將ZMQ運行起來了，讓我們回顧一下為什么我們需要ZMQ：
目前的應用程序很多都會包含跨網絡的組件，無論是局域網還是因特網。這些程序的開發者都會用到某種消息通信機制。有些人會使用某種消息隊列產品，而大多數人則會自己手工來做這些事，使用TCP或UDP協議。這些協議使用起來并不困難，但是，簡單地將消息從A發給B，和在任何情況下都能進行可靠的消息傳輸，這兩種情況顯然是不同的。
讓我們看看在使用純TCP協議進行消息傳輸時會遇到的一些典型問題。任何可復用的消息傳輸層肯定或多或少地會要解決以下問題：
* 如何處理I/O？是讓程序阻塞等待響應，還是在后臺處理這些事？這是軟件設計的關鍵因素。阻塞式的I/O操作會讓程序架構難以擴展，而后臺處理I/O也是比較困難的。
* 如何處理那些臨時的、來去自由的組件？我們是否要將組件分為客戶端和服務端兩種，并要求服務端永不消失？那如果我們想要將服務端相連怎么辦？我們要每隔幾秒就進行重連嗎？
* 我們如何表示一條消息？我們怎樣通過拆分消息，讓其變得易讀易寫，不用擔心緩存溢出，既能高效地傳輸小消息，又能勝任視頻等大型文件的傳輸？
* 如何處理那些不能立刻發送出去的消息？比如我們需要等待一個網絡組件重新連接的時候？我們是直接丟棄該條消息，還是將它存入數據庫，或是內存中的一個隊列？
* 要在哪里保存消息隊列？如果某個組件讀取消息隊列的速度很慢，造成消息的堆積怎么辦？我們要采取什么樣的策略？
* 如何處理丟失的消息？我們是等待新的數據，請求重發，還是需要建立一套新的可靠性機制以保證消息不會丟失？如果這個機制自身崩潰了呢？
* 如果我們想換一種網絡連接協議，如用廣播代替TCP單播？或者改用IPv6？我們是否需要重寫所有的應用程序，或者將這種協議抽象到一個單獨的層中？
* 我們如何對消息進行路由？我們可以將消息同時發送給多個節點嗎？是否能將應答消息返回給請求的發送方？
* 我們如何為另一種語言寫一個API？我們是否需要完全重寫某項協議，還是重新打包一個類庫？
* 怎樣才能做到在不同的架構之間傳送消息？是否需要為消息規定一種編碼？
* 我們如何處理網絡通信錯誤？等待并重試，還是直接忽略或取消？
我們可以找一個開源軟件來做例子，如[Hadoop Zookeeper](http://hadoop.apache.org/zookeeper/)，看一下它的C語言API源碼，[src/c/src/zookeeper.c]([http://github.com/apache/zookeeper/blob/trunk/src/c/src/zookeeper.c src/c/src/zookeeper.c)。這段代碼大約有3200行，沒有注釋，實現了一個C/S網絡通信協議。它工作起來很高效，因為使用了poll()來代替select()。但是，Zookeeper應該被抽象出來，作為一種通用的消息通信層，并加以詳細的注釋。像這樣的模塊應該得到最大程度上的復用，而不是重復地制造輪子。
但是，如何編寫這樣一個可復用的消息層呢？為什么長久以來人們寧愿在自己的代碼中重復書寫控制原始TCP套接字的代碼，而不愿編寫這樣一個公共庫呢？
其實，要編寫一個通用的消息層是件非常困難的事，這也是為什么FOSS項目不斷在嘗試，一些商業化的消息產品如此之復雜、昂貴、僵硬、脆弱。2006年，iMatix設計了AMQP協議，為FOSS項目的開發者提供了可能是當時第一個可復用的消息系統。[AMQP][]比其他同類產品要來得好，但[仍然是復雜、昂貴和脆弱的](http://www.imatix.com/articles:whats-wrong-with-amqp)。它需要花費幾周的時間去學習，花費數月的時間去創建一個真正能用的架構，到那時可能為時已晚了。
大多數消息系統項目，如AMQP，為了解決上面提到的種種問題，發明了一些新的概念，如“代理”的概念，將尋址、路由、隊列等功能都包含了進來。結果就是在一個沒有任何注釋的協議之上，又構建了一個C/S協議和相應的API，讓應用程序和代理相互通信。代理的確是一個不錯的解決方案，幫助降低大型網絡結構的復雜度。但是，在Zookeeper這樣的項目中應用代理機制的消息系統，可能是件更加糟糕的事，因為這意味了需要添加一臺新的計算機，并構成一個新的單點故障。代理會逐漸成為新的瓶頸，管理起來更具風險。如果軟件支持，我們可以添加第二個、第三個、第四個代理，構成某種冗余容錯的模式。有人就是這么做的，這讓系統架構變得更為復雜，增加了隱患。
在這種以代理為中心的架構下，需要一支專門的運維團隊。你需要晝夜不停地觀察代理的狀態，不時地用棍棒調教他們。你需要添加計算機，以及更多的備份機，你需要有專人管理這些機器。這樣做只對那些大型的網絡應用程序才有意義，因為他們有更多可移動的模塊，有多個團隊進行開發和維護，而且已經經過了多年的建設。
這樣一來，中小應用程序的開發者們就無計可施了。他們只能設法避免編寫網絡應用程序，轉而編寫那些不需要擴展的程序；或者可以使用原始的方式進行網絡編程，但編寫的軟件會非常脆弱和復雜，難以維護；亦或者他們選擇一種消息通信產品，雖然能夠開發出擴展性強的應用程序，但需要支付高昂的代價。似乎沒有一種選擇是合理的，這也是為什么在上個世紀消息系統會成為一個廣泛的問題。
我們真正需要的是這樣一種消息軟件，它能夠做大型消息軟件所能做的一切，但使用起來又非常簡單，成本很低，可以用到所有的應用程序中，沒有任何依賴條件。因為沒有了額外的模塊，就降低了出錯的概率。這種軟件需要能夠在所有的操作系統上運行，并能支持所有的編程語言。
ZMQ就是這樣一種軟件：它高效，提供了嵌入式的類庫，使應用程序能夠很好地在網絡中擴展，成本低廉。
ZMQ的主要特點有：
* ZMQ會在后臺線程異步地處理I/O操作，它使用一種不會死鎖的數據結構來存儲消息。
* 網絡組件可以來去自如，ZMQ會負責自動重連，這就意味著你可以以任何順序啟動組件；用它創建的面向服務架構（SOA）中，服務端可以隨意地加入或退出網絡。
* ZMQ會在有必要的情況下自動將消息放入隊列中保存，一旦建立了連接就開始發送。
* ZMQ有閾值（HWM）的機制，可以避免消息溢出。當隊列已滿，ZMQ會自動阻塞發送者，或丟棄部分消息，這些行為取決于你所使用的消息模式。
* ZMQ可以讓你用不同的通信協議進行連接，如TCP、廣播、進程內、進程間。改變通信協議時你不需要去修改代碼。
* ZMQ會恰當地處理速度較慢的節點，會根據消息模式使用不同的策略。
* ZMQ提供了多種模式進行消息路由，如請求-應答模式、發布-訂閱模式等。這些模式可以用來搭建網絡拓撲結構。
* ZMQ中可以根據消息模式建立起一些中間裝置（很小巧），可以用來降低網絡的復雜程度。
* ZMQ會發送整個消息，使用消息幀的機制來傳遞。如果你發送了10KB大小的消息，你就會收到10KB大小的消息。
* ZMQ不強制使用某種消息格式，消息可以是0字節的，或是大到GB級的數據。當你表示這些消息時，可以選用諸如谷歌的protocol buffers，XDR等序列化產品。
* ZMQ能夠智能地處理網絡錯誤，有時它會進行重試，有時會告知你某項操作發生了錯誤。
* ZMQ甚至可以降低對環境的污染，因為節省了CPU時間意味著節省了電能。
其實ZMQ可以做的還不止這些，它會顛覆人們編寫網絡應用程序的模式。雖然從表面上看，它不過是提供了一套處理套接字的API，能夠用zmq_recv()和zmq_send()進行消息的收發，但是，消息處理將成為應用程序的核心部分，很快你的程序就會變成一個個消息處理模塊，這既美觀又自然。它的擴展性還很強，每項任務由一個節點（節點是一個線程）、同一臺機器上的兩個節點（節點是一個進程）、同一網絡上的兩臺機器（節點是一臺機器）來處理，而不需要改動應用程序。
### 套接字的擴展性
我們來用實例看看ZMQ套接字的擴展性。這個腳本會啟動氣象信息服務及多個客戶端：
```
wuserver &
wuclient 12345 &
wuclient 23456 &
wuclient 34567 &
wuclient 45678 &
wuclient 56789 &
```
執行過程中，我們可以通過top命令查看進程狀態（以下是一臺四核機器的情況）：
```
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
7136 ph 20 0 1040m 959m 1156 R 157 12.0 16:25.47 wuserver
7966 ph 20 0 98608 1804 1372 S 33 0.0 0:03.94 wuclient
7963 ph 20 0 33116 1748 1372 S 14 0.0 0:00.76 wuclient
7965 ph 20 0 33116 1784 1372 S 6 0.0 0:00.47 wuclient
7964 ph 20 0 33116 1788 1372 S 5 0.0 0:00.25 wuclient
7967 ph 20 0 33072 1740 1372 S 5 0.0 0:00.35 wuclient
```
我們想想現在發生了什么：氣象信息服務程序有一個單獨的套接字，卻能同時向五個客戶端并行地發送消息。我們可以有成百上千個客戶端并行地運作，服務端看不到這些客戶端，不能操縱它們。
### 如果解決丟失消息的問題
在編寫ZMQ應用程序時，你遇到最多的問題可能是無法獲得消息。下面有一個問題解決路線圖，列舉了最基本的出錯原因。不用擔心其中的某些術語你沒有見過，在后面的幾章里都會講到。
如果ZMQ在你的應用程序中扮演非常重要的角色，那你可能就需要好好計劃一下了。首先，創建一個原型，用以測試設計方案的可行性。采取一些壓力測試的手段，確保它足夠的健壯。其次，主攻測試代碼，也就是編寫測試框架，保證有足夠的電力供應和時間，來進行高強度的測試。理想狀態下，應該由一個團隊編寫程序，另一個團隊負責擊垮它。最后，讓你的公司及時[聯系iMatix](http://www.imatix.com/contact)，獲得技術上的支持。
簡而言之，如果你沒有足夠理由說明設計出來的架構能夠在現實環境中運行，那么很有可能它就會在最緊要的關頭崩潰。
### 警告：你的想法可能會被顛覆！
傳統網絡編程的一個規則是套接字只能和一個節點建立連接。雖然也有廣播的協議，但畢竟是第三方的。當我們認定“一個套接字 = 一個連接”的時候，我們會用一些特定的方式來擴展應用程序架構：我們為每一塊邏輯創建線程，該線程獨立地維護一個套接字。
但在ZMQ的世界里，套接字是智能的、多線程的，能夠自動地維護一組完整的連接。你無法看到它們，甚至不能直接操縱這些連接。當你進行消息的收發、輪詢等操作時，只能和ZMQ套接字打交道，而不是連接本身。所以說，ZMQ世界里的連接是私有的，不對外部開放，這也是ZMQ易于擴展的原因之一。
由于你的代碼只會和某個套接字進行通信，這樣就可以處理任意多個連接，使用任意一種網絡協議。而ZMQ的消息模式又可以進行更為廉價和便捷的擴展。
這樣一來，傳統的思維就無法在ZMQ的世界里應用了。在你閱讀示例程序代碼的時候，也許你腦子里會想方設法地將這些代碼和傳統的網絡編程相關聯：當你讀到“套接字”的時候，會認為它就表示與另一個節點的連接——這種想法是錯誤的；當你讀到“線程”時，會認為它是與另一個節點的連接——這也是錯誤的。
如果你是第一次閱讀本指南，使用ZMQ進行了一兩天的開發（或者更長），可能會覺得疑惑，ZMQ怎么會讓事情便得如此簡單。你再次嘗試用以往的思維去理解ZMQ，但又無功而返。最后，你會被ZMQ的理念所折服，撥云見霧，開始享受ZMQ帶來的樂趣。
[iMatix]: http://www.imatix.com/
[AMQP]: http://www.amqp.org/
