关于消息队列——ZeroMQ的安装

ZeroMQ —— The Intelligent Transport Layer (智能的传输层封装)

1,与其说它可嵌入的网络库，不如说是一个并发的socket框架
2,集群上使用比TCP快速
3,消息通过inpro，IPC，TCP以及广播方式传输
4,通过fanout(扇出)，pubsub，pipeline，request-reply实现多对多通信
5,异步IO
6,活跃社区支持
7,拥有30多种语言的扩展库
8,支持window，linux，os x等系统
9,LGPL许可

 

安装过程如下：

下载源码zeromq-3.2.2.tar.gz编译安装，然后安装php扩展

git clone git://github.com/mkoppanen/php-zmq.git
phpize
./configure --with-php-config=/root/bin/php-config
make && make install

编辑php.ini 添加 extension=zmq.so

 

编程参考手册 http://php.zero.mq

这里有大量程序示例可供参考https://github.com/imatix/zguide

 

A，request/reply

 

写一个服务端 server.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);

$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");
}

写一个客户端client.php

<?php

//allocate a new context
$context = new ZMQContext();

//create a new socket
$queue = $context->getSocket(ZMQ::SOCKET_REQ,"mysock");

//connect the server
$queue->connect("tcp://127.0.0.1:5555");

//send and receive message
var_dump($queue->send("Hello, using mysock")->recv());

?>

 一个简单的服务应答就搭建好了，支持并发。可以开启多个client.php去连接服务器端测试。这是简单的request-reply，下面看看如何实现pub/sub（发布/订阅）

 

B，publish/subscribe

 

发布端 publisher.php

<?php

// Prepare our context and publisher
$context = new ZMQContext();
$publisher = $context->getSocket(ZMQ::SOCKET_PUB);
$publisher->bind("tcp://127.0.0.1:5556");

while (true) {

    $filter = mt_rand(1, 9);

    // Send message to all subscribers
    $update = sprintf ("%d Hello%d", $filter, $filter);
    $publisher->send($update);
    usleep(100000);
}

?>

订阅端 subscriber.php

<?php

$context = new ZMQContext();

// Socket to talk to server
echo "Subscribe message from server...", PHP_EOL;
$subscriber = new ZMQSocket($context, ZMQ::SOCKET_SUB);
$subscriber->connect("tcp://127.0.0.1:5556");

// Subscribe to 1 or other channel
$filter = $_SERVER['argc'] > 1 ? $_SERVER['argv'][1] : "1";
$subscriber->setSockOpt(ZMQ::SOCKOPT_SUBSCRIBE, $filter);

// Process 100 updates
$total_temp = 0;
while(true) {
    $string = $subscriber->recv();
    sscanf ($string, "%d %s", $filter, $msg);
    echo "$filter recv $msg\n";
}

?>

 运行一个publisher.php端，然后启动多个subscriber.php
a, php subscriber.php 1 (能收到$filter=1的所有消息)
b, php subscriber.php 1 (能收到$filter=1的所有消息)
c, php subscriber.php 3 (能收到$filter=3的所有消息)