ZeroMQ，史上最快的消息队列
—– ZMQ的学习和研究
一、ZeroMQ的背景介绍
引用官方的说法： “ZMQ(以下ZeroMQ简称ZMQ)是一个简单好用的传输层，像框架一样的一个socket library，他使得Socket编程更加简单、简洁和性能更高。是一个消息处理队列库，可在多个线程、内核和主机盒之间弹性伸缩。ZMQ的明确目标是“成为标准网络协议栈的一部分，之后进入Linux内核”。现在还未看到它们的成功。但是，它无疑是极具前景的、并且是人们更加需要的“传统”BSD套接字之上的一 层封装。ZMQ让编写高性能网络应用程序极为简单和有趣。”
近几年有关”Message Queue”的项目层出不穷，知名的就有十几种，这主要是因为后摩尔定律时代，分布式处理逐渐成为主流，业界需要一套标准来解决分布式计算环境中节点之间的消息通信。几年的竞争下来，Apache基金会旗下的符合AMQP/1.0标准的RabbitMQ已经得到了广泛的认可，成为领先的MQ项目。
与RabbitMQ相比，ZMQ并不像是一个传统意义上的消息队列服务器，事实上，它也根本不是一个服务器，它更像是一个底层的网络通讯库，在Socket API之上做了一层封装，将网络通讯、进程通讯和线程通讯抽象为统一的API接口。
二、ZMQ是什么？
阅读了ZMQ的Guide文档后，我的理解是，这是个类似于Socket的一系列接口，他跟Socket的区别是：普通的socket是端到端的（1:1的关系），而ZMQ却是可以N：M 的关系，人们对BSD套接字的了解较多的是点对点的连接，点对点连接需要显式地建立连接、销毁连接、选择协议（TCP/UDP）和处理错误等，而ZMQ屏蔽了这些细节，让你的网络编程更为简单。ZMQ用于node与node间的通信，node可以是主机或者是进程。
三、本文的目的
在集群对外提供服务的过程中，我们有很多的配置，需要根据需要随时更新，那么这个信息如果推动到各个节点？并且保证信息的一致性和可靠性？本文在介绍ZMQ基本理论的基础上，试图使用ZMQ实现一个配置分发中心。从一个节点，将信息无误的分发到各个服务器节点上，并保证信息正确性和一致性。
四、ZMQ的三个基本模型
ZMQ提供了三个基本的通信模型，分别是“Request-Reply “，”Publisher-Subscriber“，”Parallel Pipeline”,我们从这三种模式一窥ZMQ的究竟
ZMQ的hello world！
由Client发起请求，并等待Server回应请求。请求端发送一个简单的hello，服务端则回应一个world。请求端和服务端都可以是 1:N 的模型。通常把 1 认为是 Server ，N 认为是Client 。ZMQ 可以很好的支持路由功能（实现路由功能的组件叫作Device），把 1:N 扩展为N:M （只需要加入若干路由节点）。如图1所示：
图1：ZMQ的Request-Reply 通信
服务端的php程序如下：
代碼: 選擇全部
<?php
/*
* Hello World server
* Binds REP socket to tcp://*:5555
* Expects "Hello" from client, replies with "World"
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext(1);
// Socket to talk to clients
$responder = new ZMQSocket($context, ZMQ::SOCKET_REP);
$responder->bind("tcp://*:5555");
while(true) {
// Wait for next request from client
$request = $responder->recv();
printf ("Received request: [%s]\n", $request);
// Do some 'work'
sleep (1);
// Send reply back to client
$responder->send("World");
}
代碼: 選擇全部
<?php
/*
* Hello World client
* Connects REQ socket to tcp://localhost:5555
* Sends "Hello" to server, expects "World" back
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext();
// Socket to talk to server
echo "Connecting to hello world server...\n";
$requester = new ZMQSocket($context, ZMQ::SOCKET_REQ);
$requester->connect("tcp://localhost:5555");
for($request_nbr = 0; $request_nbr != 10; $request_nbr++) {
printf ("Sending request %d...\n", $request_nbr);
$requester->send("Hello");
$reply = $requester->recv();
printf ("Received reply %d: [%s]\n", $request_nbr, $reply);
}
a) 服务端和客户端无论谁先启动，效果是相同的，这点不同于Socket。
b) 在服务端收到信息以前，程序是阻塞的，会一直等待客户端连接上来。
c) 服务端收到信息以后，会send一个“World”给客户端。值得注意的是一定是client连接上来以后，send消息给Server，然后Server再rev然后响应client，这种一问一答式的。如果Server先send，client先rev是会报错的。
d) ZMQ通信通信单元是消息，他除了知道Bytes的大小，他并不关心的消息格式。因此，你可以使用任何你觉得好用的数据格式。Xml、Protocol Buffers、Thrift、json等等。
e) 虽然可以使用ZMQ实现HTTP协议，但是，这绝不是他所擅长的。
ZMQ的Publish-subscribe模式
我们可以想象一下天气预报的订阅模式，由一个节点提供信息源，由其他的节点，接受信息源的信息，如图2所示：
图2：ZMQ的Publish-subscribe
示例代码如下 ：
Publisher：
代碼: 選擇全部
<?php
/*
* Weather update server
* Binds PUB socket to tcp://*:5556
* Publishes random weather updates
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
// Prepare our context and publisher
$context = new ZMQContext();
$publisher = $context->getSocket(ZMQ::SOCKET_PUB);
$publisher->bind("tcp://*:5556");
while (true) {
// Get values that will fool the boss
$zipcode = mt_rand(0, 100000);
$temperature = mt_rand(-80, 135);
$relhumidity = mt_rand(10, 60);
// Send message to all subscribers
$update = sprintf ("%05d %d %d", $zipcode, $temperature, $relhumidity);
$publisher->send($update);
}</pre>
Subscriber
<pre><?php
/*
* Weather update client
* Connects SUB socket to tcp://localhost:5556
* Collects weather updates and finds avg temp in zipcode
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext();
// Socket to talk to server
echo "Collecting updates from weather server…", PHP_EOL;
$subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);
$subscriber->connect("tcp://localhost:5556");
// Subscribe to zipcode, default is NYC, 10001
$filter = $_SERVER['argc'] > 1 ? $_SERVER['argv'][1] : "10001";
$subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, $filter);
// Process 100 updates
$total_temp = 0;
for ($update_nbr = 0; $update_nbr < 100; $update_nbr++) {
$string = $subscriber->recv();
sscanf ($string, "%d %d %d", $zipcode, $temperature, $relhumidity);
$total_temp += $temperature;
}
printf ("Average temperature for zipcode '%s' was %dF\n",
$filter, (int) ($total_temp / $update_nbr));
a) 与Hello World不同的是，Socket的类型变成SOCKET_PUB和SOCKET_SUB类型。
b) 客户端需要$subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, $filter);设置一个过滤值，相当于设定一个订阅频道，否则什么信息也收不到。
c) 服务器端一直不断的广播中，如果中途有Subscriber端退出，并不影响他继续的广播，当Subscriber再连接上来的时候，收到的就是后来发送的新的信息了。这对比较晚加入的，或者是中途离开的订阅者，必然会丢失掉一部分信息，这是这个模式的一个问题，所谓的Slow joiner。稍后，会解决这个问题。
d) 但是，如果Publisher中途离开，所有的Subscriber会hold住，等待Publisher再上线的时候，会继续接受信息。
ZMQ的PipeLine模型
想象一下这样的场景，如果需要统计各个机器的日志，我们需要将统计任务分发到各个节点机器上，最后收集统计结果，做一个汇总。PipeLine比较适合于这种场景，他的结构图，如图3所示。
图3：ZMQ的PipeLine模型
Parallel task ventilator in PHP
代碼: 選擇全部
<?php
/*
* Task ventilator
* Binds PUSH socket to tcp://localhost:5557
* Sends batch of tasks to workers via that socket
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext();
// Socket to send messages on
$sender = new ZMQSocket($context, ZMQ::SOCKET_PUSH);
$sender->bind("tcp://*:5557");
echo "Press Enter when the workers are ready: ";
$fp = fopen('php://stdin', 'r');
$line = fgets($fp, 512);
fclose($fp);
echo "Sending tasks to workers…", PHP_EOL;
// The first message is "0" and signals start of batch
$sender->send(0);
// Send 100 tasks
$total_msec = 0; // Total expected cost in msecs
for ($task_nbr = 0; $task_nbr < 100; $task_nbr++) {
// Random workload from 1 to 100msecs
$workload = mt_rand(1, 100);
$total_msec += $workload;
$sender->send($workload);
}
printf ("Total expected cost: %d msec\n", $total_msec);
sleep (1); // Give 0MQ time to deliver
代碼: 選擇全部
<?php
/*
* Task worker
* Connects PULL socket to tcp://localhost:5557
* Collects workloads from ventilator via that socket
* Connects PUSH socket to tcp://localhost:5558
* Sends results to sink via that socket
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
$context = new ZMQContext();
// Socket to receive messages on
$receiver = new ZMQSocket($context, ZMQ::SOCKET_PULL);
$receiver->connect("tcp://localhost:5557");
// Socket to send messages to
$sender = new ZMQSocket($context, ZMQ::SOCKET_PUSH);
$sender->connect("tcp://localhost:5558");
// Process tasks forever
while (true) {
$string = $receiver->recv();
// Simple progress indicator for the viewer
echo $string, PHP_EOL;
// Do the work
usleep($string * 1000);
// Send results to sink
$sender->send("");
}
代碼: 選擇全部
<?php
/*
* Task sink
* Binds PULL socket to tcp://localhost:5558
* Collects results from workers via that socket
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
// Prepare our context and socket
$context = new ZMQContext();
$receiver = new ZMQSocket($context, ZMQ::SOCKET_PULL);
$receiver->bind("tcp://*:5558");
// Wait for start of batch
$string = $receiver->recv();
// Start our clock now
$tstart = microtime(true);
// Process 100 confirmations
$total_msec = 0; // Total calculated cost in msecs
for ($task_nbr = 0; $task_nbr < 100; $task_nbr++) {
$string = $receiver->recv();
if($task_nbr % 10 == 0) {
echo ":";
} else {
echo ".";
}
}
$tend = microtime(true);
$total_msec = ($tend - $tstart) * 1000;
echo PHP_EOL;
printf ("Total elapsed time: %d msec", $total_msec);
echo PHP_EOL;
五、其他扩展模式
通常，一个节点，即可以作为Server，同时也能作为Client，通过PipeLine模型中的Worker，他向上连接着任务分发，向下连接着结果搜集的Sink机器。因此，我们可以借助这种特性，丰富的扩展原有的三种模式。例如，一个代理Publisher，作为一个内网的Subscriber接受信息，同时将信息，转发到外网，其结构图如图4所示。
图4：ZMQ的扩展模式
六、多个服务器
ZMQ和Socket的区别在于，前者支持N：M的连接，而后者则只是1：1的连接，那么一个Client连接多个Server的情况是怎样的呢，我们通过图5来说明。
图5：ZMQ的N:1的连接情况
我们假设Client有R1,R2,R3,R4四个任务，我们只需要一个ZMQ的Socket，就可以连接四个服务，他能够自动均衡的分配任务。如图5所示，R1，R4自动分配到了节点A，R2到了B，R3到了C。如果我们是N:M的情况呢？这个扩展起来，也不难，如图6所示。
图6：N:M的连接
我们通过一个中间结点（Broker）来进行负载均衡的功能。我们通过代码了解，其中的Client和我们的Hello World的Client端是一样的，而Server端的不同是，他不需要监听端口，而是需要连接Broker的端口，接受需要处理的信息。所以，我们重点阅读Broker的代码：
代碼: 選擇全部
<?php
/*
* Simple request-reply broker
* @author Ian Barber <ian(dot)barber(at)gmail(dot)com>
*/
// Prepare our context and sockets
$context = new ZMQContext();
$frontend = new ZMQSocket($context, ZMQ::SOCKET_ROUTER);
$backend = new ZMQSocket($context, ZMQ::SOCKET_DEALER);
$frontend->bind("tcp://*:5559");
$backend->bind("tcp://*:5560");
// Initialize poll set
$poll = new ZMQPoll();
$poll->add($frontend, ZMQ::POLL_IN);
$poll->add($backend, ZMQ::POLL_IN);
$readable = $writeable = array();
// Switch messages between sockets
while(true) {
$events = $poll->poll($readable, $writeable);
foreach($readable as $socket) {
if($socket === $frontend) {
// Process all parts of the message
while(true) {
$message = $socket->recv();
// Multipart detection
$more = $socket->getSockOpt(ZMQ::SOCKOPT_RCVMORE);
$backend->send($message, $more ? ZMQ::MODE_SNDMORE : null);
if(!$more) {
break; // Last message part
}
}
}
else if($socket === $backend) {
$message = $socket->recv();
// Multipart detection
$more = $socket->getSockOpt(ZMQ::SOCKOPT_RCVMORE);
$frontend->send($message, $more ? ZMQ::MODE_SNDMORE : null);
if(!$more) {
break; // Last message part
}
}
}
}
七、进程间的通信
ZMQ不仅能通过TCP完成节点间的通信，也可以通过Socket文件完成进程间的通信。如图7所示，我们fork三个PHP进程，将进程1的数据，通过Socket文件发送到进程3。
图7：进程间的通信
代碼: 選擇全部
<?php
function step1() {
$context = new ZMQContext();
// Signal downstream to step 2
$sender = new ZMQSocket($context, ZMQ::SOCKET_PAIR);
$sender->connect("ipc://step2.ipc");
$sender->send("hello ,i am step1");
}
function step2() {
$pid = pcntl_fork();
if($pid == 0) {
step1();
exit();
}
$context = new ZMQContext();
// Bind to ipc: endpoint, then start upstream thread
$receiver = new ZMQSocket($context, ZMQ::SOCKET_PAIR);
$receiver->bind("ipc://step2.ipc");
// Wait for signal
sleep(10);
$strings = $receiver->recv();
echo "step2 receiver is $strings". PHP_EOL;
sleep(10);
// Signal downstream to step 3
$sender = new ZMQSocket($context, ZMQ::SOCKET_PAIR);
$sender->connect("ipc://step3.ipc");
$sender->send($strings);
}
// Start upstream thread then bind to icp: endpoint
$pid = pcntl_fork();
if($pid == 0) {
step2();
exit();
}
$context = new ZMQContext();
$receiver = new ZMQSocket($context, ZMQ::SOCKET_PAIR);
$receiver->bind("ipc://step3.ipc");
// Wait for signal
$sr = $receiver->recv();
echo "the result is {$sr}".PHP_EOL;
图8：运行过程中生成的文件
八、利用ZeroMQ实现一个配置推送中心
当我们将WEB代码部署到集群上的时候，如果需要实时的将最新的配置信息，主动的推送到各个机器节点。在此过程中，我们一定要保证，各个节点收到的信息的一致性和正确性，如果使用HTTP，由于他的无状态性，我们无法保证信息的一致性，当然，你可以使用HTTP来实现，只是更复杂，为什么不用ZMQ？他能让你更简单的实现这些功能。
我们使用ZMQ的信息订阅模式。在那个模式中，我们注意到，对于后来的加入节点，始终会丢失在他加入之前，已经发送的信息（Slow joiner）。我们可以开启另外一个ZMQ的通信通道，用于报告当前节点的情况（节点的身份、准备状态等），其结构如图9所示。
图9：扩展ZMQ的订阅者模式
我们通过$context->getSocket(ZMQ::SOCKET_REQ);设置一个新的Request-Reply连接，来用于Subscriber向Publisher报告自己的身份信息，而Publisher则等待所有的Subscriber都连接上的时候，再选择Publish自己的信息。
Subscriber端的程序如下：
代碼: 選擇全部
<?php
$hostname = $_SERVER['argc'] > 1 ? $_SERVER['argv'][1] : "s1";
$context = new ZMQContext(2);
$sub = new ZMQSocket($context,ZMQ::SOCKET_SUB);
$sub->connect("tcp://localhost:5561");
//$subscriber->setSockOpt(ZMQ::SOCKOPT_IDENTITY, $hostname);
$sub->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE,"");
$client = $context->getSocket(ZMQ::SOCKET_REQ);
$client->connect("tcp://localhost:5562");
while(1) {
//$client->connect("tcp://localhost:5562");
$client->send($hostname);
$version = $client->recv();
echo $version."\r\n";
if (!empty($version)) {
$recive = $sub->recv();
$vars = json_decode($recive);
var_dump($vars);
}
}
代碼: 選擇全部
<?php
$CONFIG["TAOKE_BTS"]["ENABLE"] = true;
$CONFIG["QP_BTS"]["ENABLE"] = true;
$CONFIG["QP_BTS"]["TK_TEST"] = 13;
$string = json_encode($CONFIG);
$clients = array("s2","s1","s3");
$context = new ZMQContext(10);
//Socket talk to clients
$publisher = new ZMQSocket($context,ZMQ::SOCKET_PUB);
$publisher->bind("tcp://*:5561");
//Socket to publish message
$server = new ZMQSocket($context,ZMQ::SOCKET_REP);
$server->bind("tcp://*:5562");
while(count($clients)!=0) {
$client_name = $server->recv();
echo "{$client_name} is connect!\r\n";
if (in_array($client_name, $clients)) { //coming one client
$key = array_search($client_name, $clients);
unset($clients[$key]);
echo "$client_name has come in!\r\n";
$server->send("Version is 2.0");
} else {
$server->send("You are a stranger!");
}
}
$publisher->send($string);
?>
后续的处理，Subscriber可以选择将配置信息写入到APC缓存，程序将始终从缓存中读取部分配置信息，Subscriber并将更新后的状态信息，实时的通过5562报告给Publisher。
虽然，在本示例中不会出现，但是，如果需要发布的信息量过大，在接受信息的过程中，Subscriber端突然中断网络（或者是程序崩溃），那么当他在连接上来的时候，有部分信息就会丢失？ZMQ考虑到这个问题，通过$subscriber->setSockOpt(ZMQ::SOCKOPT_IDENTITY, $hostname);设置一个id，当这个id的Subscriber重新连接上来的时候，他可以从上次中断的地方，继续接受信息，当然，节点的中断，不会影响其他的节点继续的接受信息。
那么ZMQ是怎么实现断线重连后，继续发送信息呢 ？他会将断开的Subscriber应该接受到的信息发到内存中，等待他重新上线后，将缓存的信息，继续发送给他。当然，内存必然是有限的，过多就会出现内存溢出。ZMQ通过
SetSockOpt(ZMQ::SOCKOPT_SWAP, 250000)设置Swap空间的大小，来防止out of memory and crash。最终，我们的程序运行结果，如图10所示。
图10：配置中心的运行结果
当然，这只是一个大体的思路，如果应用到实际的成产环境中，还需要考虑更多的问题，包含稳定性，容错等等。然而，ZMQ由于高并发，以及稳定性和易用性，前景不错，他的目标是进入Linux内核，我们期待那一天的到来。
参考资料 ：
http://www.infoq.com/cn/news/2010/09/in ... on-zero-mq Infoq对zeromq的简介
http://zguide.zeromq.org/page:all ZeroMQ的guide文档