电子商务网站建设 论文,教育网站模板下载,郑州电力高等专科学校在哪个区,wordpress装机主题一、ZeroMQ 的背景介绍 引用官方的说法#xff1a; “ZMQ (以下 ZeroMQ 简称 ZMQ)是一个简单好用的传输层#xff0c;像框架一样的一个 socket library#xff0c;他使得 Socket 编程更加简单、简洁和性能更高。是一个消息处理队列库#xff0c;可在多个线程、内核和主机盒… 一、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 却是可以NM 的关系人们对 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 所示 图1ZMQ 的 Request-Reply 通信 服务端的 php 程序如下 ?php/** Hello World server* Binds REP socket to tcp://*:5555* Expects Hello from client, replies with World* author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt;*/$context new ZMQContext (1);// Socket to talk to clients$responder new ZMQSocket ($context, ZMQ::SOCKET_REP);$responder-gt;bind (tcp://*:5555);while(true) {// Wait for next request from client$request $responder-gt;recv ();printf (Received request: [%s]\n, $request); // Do some worksleep (1); // Send reply back to client$responder-gt;send (World);} Client 程序如下 ?php /* * Hello World client * Connects REQ socket to tcp://localhost:5555 * Sends Hello to server, expects World back * author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt; */$context new ZMQContext (); // Socket to talk to server echo Connecting to hello world server...\n; $requester new ZMQSocket ($context, ZMQ::SOCKET_REQ); $requester-gt;connect (tcp://localhost:5555); for($request_nbr 0; $request_nbr ! 10; $request_nbr) { printf (Sending request %d...\n, $request_nbr); $requester-gt;send (Hello); $reply $requester-gt;recv (); printf (Received reply %d: [%s]\n, $request_nbr, $reply); } 从以上的过程我们可以了解到使用 ZMQ 写基本的程序的方法需要注意的是 a) 服务端和客户端无论谁先启动效果是相同的这点不同于 Socket。 b) 在服务端收到信息以前程序是阻塞的会一直等待客户端连接上来。 c) 服务端收到信息以后会 send 一个“World”给客户端。值得注意的是一定是 client 连接上来以后send 消息给 Server然后 Server 再 rev 然后响应 client这种一问一答式的。如果 Server 先 sendclient 先 rev 是会报错的。 d) ZMQ 通信通信单元是消息他除了知道 Bytes 的大小他并不关心的消息格式。因此你可以使用任何你觉得好用的数据格式。Xml、Protocol Buffers、Thrift、json 等等。 e) 虽然可以使用 ZMQ 实现 HTTP 协议但是这绝不是他所擅长的。 ZMQ 的 Publish-subscribe 模式 我们可以想象一下天气预报的订阅模式由一个节点提供信息源由其他的节点接受信息源的信息如图 2 所示 图2ZMQ 的 Publish-subscribe 示例代码如下 Publisher ?php/** Weather update server* Binds PUB socket to tcp://*:5556* Publishes random weather updates* author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt;*/ // Prepare our context and publisher$context new ZMQContext ();$publisher $context-gt;getSocket (ZMQ::SOCKET_PUB);$publisher-gt;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-gt;send ($update);}/preSubscriberprelt;?php/** Weather update client* Connects SUB socket to tcp://localhost:5556* Collects weather updates and finds avg temp in zipcode* author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt;*/ $context new ZMQContext (); // Socket to talk to serverecho Collecting updates from weather server…, PHP_EOL;$subscriber new ZMQSocket ($context, ZMQ::SOCKET_SUB);$subscriber-gt;connect (tcp://localhost:5556); // Subscribe to zipcode, default is NYC, 10001$filter $_SERVER[argc] gt; 1 ? $_SERVER[argv][1] : 10001;$subscriber-gt;setSockOpt (ZMQ::SOCKOPT_SUBSCRIBE, $filter); // Process 100 updates$total_temp 0;for ($update_nbr 0; $update_nbr lt; 100; $update_nbr) {$string $subscriber-gt;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)); 这段代码讲的是服务器端生成随机数 zipcode、temperature、relhumidity 分别代表城市代码、温度值和湿度值。然后不断的广播信息而客户端通过设置过滤参数接受特定城市代码的信息收集完了以后做一个平均值。 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 所示。 图3ZMQ 的 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 lt;ian (dot) barber (at) gmail (dot) comgt;*/ $context new ZMQContext (); // Socket to send messages on$sender new ZMQSocket ($context, ZMQ::SOCKET_PUSH);$sender-gt;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-gt;send (0); // Send 100 tasks$total_msec 0; // Total expected cost in msecsfor ($task_nbr 0; $task_nbr lt; 100; $task_nbr) {// Random workload from 1 to 100msecs$workload mt_rand(1, 100);$total_msec $workload;$sender-gt;send ($workload); }printf (Total expected cost: %d msec\n, $total_msec);sleep (1); // Give 0MQ time to deliver Parallel task worker in PHP ?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 lt;ian (dot) barber (at) gmail (dot) comgt;*/ $context new ZMQContext (); // Socket to receive messages on$receiver new ZMQSocket ($context, ZMQ::SOCKET_PULL);$receiver-gt;connect (tcp://localhost:5557); // Socket to send messages to$sender new ZMQSocket ($context, ZMQ::SOCKET_PUSH);$sender-gt;connect (tcp://localhost:5558); // Process tasks foreverwhile (true) {$string $receiver-gt;recv (); // Simple progress indicator for the viewerecho $string, PHP_EOL; // Do the workusleep($string * 1000); // Send results to sink$sender-gt;send ();} Parallel task sink in PHP ?php/** Task sink* Binds PULL socket to tcp://localhost:5558* Collects results from workers via that socket* author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt;*/ // Prepare our context and socket$context new ZMQContext ();$receiver new ZMQSocket ($context, ZMQ::SOCKET_PULL);$receiver-gt;bind (tcp://*:5558); // Wait for start of batch$string $receiver-gt;recv (); // Start our clock now$tstart microtime(true); // Process 100 confirmations$total_msec 0; // Total calculated cost in msecsfor ($task_nbr 0; $task_nbr lt; 100; $task_nbr) {$string $receiver-gt;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; 从程序中我们可以看到task ventilator 使用的是 SOCKET_PUSH将任务分发到 Worker 节点上。而 Worker 节点上使用 SOCKET_PULL 从上游接受任务并使用 SOCKET_PUSH 将结果汇集到 Slink。值得注意的是任务的分发的时候也同样有一个负载均衡的路由功能worker 可以随时自由加入task ventilator 可以均衡将任务分发出去。 五、其他扩展模式 通常一个节点即可以作为 Server同时也能作为 Client通过 PipeLine 模型中的 Worker他向上连接着任务分发向下连接着结果搜集的 Sink 机器。因此我们可以借助这种特性丰富的扩展原有的三种模式。例如一个代理 Publisher作为一个内网的 Subscriber 接受信息同时将信息转发到外网其结构图如图 4 所示。 图4ZMQ 的扩展模式 六、多个服务器 ZMQ 和 Socket 的区别在于前者支持NM的连接而后者则只是11的连接那么一个 Client 连接多个 Server 的情况是怎样的呢我们通过图 5 来说明。 图5ZMQ 的N:1的连接情况 我们假设 Client 有 R1,R2,R3,R4四个任务我们只需要一个 ZMQ 的 Socket就可以连接四个服务他能够自动均衡的分配任务。如图 5 所示R1R4自动分配到了节点AR2到了BR3到了C。如果我们是N:M的情况呢这个扩展起来也不难如图 6 所示。 图6N:M的连接 我们通过一个中间结点Broker来进行负载均衡的功能。我们通过代码了解其中的 Client 和我们的 Hello World 的 Client 端是一样的而 Server 端的不同是他不需要监听端口而是需要连接 Broker 的端口接受需要处理的信息。所以我们重点阅读 Broker 的代码 ?php/** Simple request-reply broker* author Ian Barber lt;ian (dot) barber (at) gmail (dot) comgt;*/ // Prepare our context and sockets$context new ZMQContext ();$frontend new ZMQSocket ($context, ZMQ::SOCKET_ROUTER);$backend new ZMQSocket ($context, ZMQ::SOCKET_DEALER);$frontend-gt;bind (tcp://*:5559);$backend-gt;bind (tcp://*:5560); // Initialize poll set$poll new ZMQPoll ();$poll-gt;add ($frontend, ZMQ::POLL_IN);$poll-gt;add ($backend, ZMQ::POLL_IN);$readable $writeable array(); // Switch messages between socketswhile(true) {$events $poll-gt;poll ($readable, $writeable); foreach($readable as $socket) {if($socket $frontend) {// Process all parts of the messagewhile(true) {$message $socket-gt;recv ();// Multipart detection$more $socket-gt;getSockOpt (ZMQ::SOCKOPT_RCVMORE);$backend-gt;send ($message, $more ? ZMQ::MODE_SNDMORE : null);if(!$more) {break; // Last message part}}}else if($socket $backend) {$message $socket-gt;recv ();// Multipart detection$more $socket-gt;getSockOpt (ZMQ::SOCKOPT_RCVMORE);$frontend-gt;send ($message, $more ? ZMQ::MODE_SNDMORE : null);if(!$more) {break; // Last message part}}}} Broker 监听了两个端口接受从多个 Client 端发送过来的数据并将数据转发给 Server。在 Broker 中我们监听了两个端口使用了两个 Socket那么对于多个 Socket 的情况我们是不需要通过轮询的方式去处理数据的在之前我们可以使用 libevent 实现异步的信息处理和传输。而现在我们只需要使用 ZMQ 的$poll-poll 以实现多个 Socket 的异步处理。 七、进程间的通信 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-gt;connect (ipc://step2.ipc); $sender-gt;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-gt;bind (ipc://step2.ipc); // Wait for signal sleep(10); $strings $receiver-gt;recv (); echo step2 receiver is $strings. PHP_EOL; sleep(10); // Signal downstream to step 3 $sender new ZMQSocket ($context, ZMQ::SOCKET_PAIR); $sender-gt;connect (ipc://step3.ipc); $sender-gt;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-gt;bind (ipc://step3.ipc); // Wait for signal $sr $receiver-gt;recv (); echo the result is {$sr}.PHP_EOL; 在运行中我们可以看到多了两个文件如图 8 所示。 图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] gt; 1 ? $_SERVER[argv][1] : s1; $context new ZMQContext (2); $sub new ZMQSocket ($context,ZMQ::SOCKET_SUB); $sub-gt;connect (tcp://localhost:5561); //$subscriber-gt;setSockOpt (ZMQ::SOCKOPT_IDENTITY, $hostname); $sub-gt;setSockOpt (ZMQ::SOCKOPT_SUBSCRIBE,); $client $context-gt;getSocket (ZMQ::SOCKET_REQ); $client-gt;connect (tcp://localhost:5562); while(1) { //$client-gt;connect (tcp://localhost:5562); $client-gt;send ($hostname); $version $client-gt;recv (); echo $version.\r\n; if (!empty($version)) { $recive $sub-gt;recv (); $vars json_decode ($recive); var_dump($vars); } } Publisher 端的程序如下 ?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-gt;bind (tcp://*:5561); //Socket to publish message $server new ZMQSocket ($context,ZMQ::SOCKET_REP); $server-gt;bind (tcp://*:5562); while(count($clients)!0) { $client_name $server-gt;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-gt;send (Version is 2.0); } else { $server-gt;send (You are a stranger!); } } $publisher-gt;send ($string); ?gt; 每个节点通过 5562 端口使用 Rep 模式和 Publisher 连接通过这个连接告之 Publisher 自己的机器名而 Publisher 端通过白名单的方式维护一个机器列表当机器列表中所有的机器连接上来以后通过 5561 端口将最新的配置信息发送出去。 后续的处理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/introduction-zero-mq Infoq 对 zeromq 的简介 http://zguide.zeromq.org/page:all ZeroMQ 的 guide 文档
