(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

原文:(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

本文将介绍在PHP中如何使用RabbitMQ来实现消息的订阅和发布。我使用的系统依然是Centos7,为了方便,应用服务器我使用Docker进行部署,容器环境:centos7+nginx+php5.6。

运行环境,安装AMQP扩展:

如何安装Docker我就不说了,网上很多教程非常简单,如果有现成的php环境可以直接使用。Docker中我使用的镜像名为webdevops/php-nginx,tag为:centos-7-php56。下载镜像:

(国际带宽出口不稳定,可能会下载失败,重试记次就好了)

docker pull webdevops/php-nginx:centos-7-php56  //下载镜像
docker run -d -p 80:80 --name rabbitmq webdevops/php-nginx:centos-7-php56 //运行容器
docker exec -ti rabbitmq /bin/bash //进入容器

进入到容器后检测下环境是否有相应扩展

cd app
vi index.php
<?php
phpinfo();

刚刚我们在运行容器的时候使用80端口,在浏览器中输入http://ip

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

搜索下没有amqp相关的信息。下面开始安装amqp扩展。

yum install gcc librabbitmq-devel.x86_64 php56w-devel
-y
wget http://pecl.php.net/get/amqp-1.4.0.tgz
tar -zxvf amqp-1.4.0.tgz
cd amqp-1.4.0
phpize
./configure --with-amqp
make && make install

在php.ini中开启extension=amqp.so 接着重启php-fpm 或 Web服务器

vi /etc/php.ini
extension=amqp.so

我这里就直接重启容器了,如果是宿主机直接安装php环境直接重启环境。

exit  //退出容器
docker restart rabbitmq //重启容器

再查看phpinfo,amqp扩展已经安装好了:

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

publish发布消息

在/app路径下新建一个publish.php的文件

touch publish.php
vi publish.php

以下是PHP代码,我们先定义好用来发消息的交换机、队列、RoutingKey、消息等变量。

$queueName = 'superrd';
$exchangeName = 'superrd';
$routeKey = 'superrd';
$message = 'Hello World!';

按照我们第二章讲到的首先建立一个连接。

$connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd'));
$connection->connect() or die("Cannot connect to the broker!\n");

新建一个信道。

$channel = new AMQPChannel($connection);

新建一个交换机Exchange,并定义属性,第二章我们讲过有四种类型的交换机,这里使用直连型DIRECT。AMQP_DURABLE代表这是一个持久化的交换机,不会以为服务器异常等因素丢失。

$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();

新建一个队列Queue,前面也讲过生产者将消息发送到Exchange中,Exchange会根据绑定关系投递到队列,也就是如果生产者在生产消息时没有队列与之绑定消息就会丢失。为了保证系统更加健硕,一般无论是消息的生产者还是消费者都会新建一遍Exchange和Queue,新建后属性不会改变。同样AMQP_DURABLE代表这是一个持久化的队列,队列会被写入磁盘。需要注意的是虽然消息是缓存在队列中,但是并不是队列是持久化的队列队列中的消息就是持久化的,消息的持久化需要单独设置。

$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();

通过routeKey绑定交换机和队列。

$queue->bind($exchangeName, $routeKey);

好了,下面可以发送消息了

$exchange->publish($message,$routeKey);

如果你希望消息也是持久化的可以使用如下的代码,实际测试结果在持久化消息后消息发布的性能下降一倍,我的磁盘是pcie的固态硬盘,如果你是机械磁盘这个性能下降估计会更明显,24核心CPU,48GB内存,pcie固态硬盘,单线程的情况下每秒可以发布2.5万左右的非持久化消息,持久化之后变为变为1.2万左右。

$exchange->publish($message,$routeKey,AMQP_NOPARAM, array('delivery_mode'=>2));

断开连接。

$connection->disconnect();

同样在发布消息之后可以通过WEB工具来查看是否发布成功,

查看交换机多了一个superid交换机。

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

查看交换机已经有superrd队列。

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

点击队列查看队列详情。Bindings标签可以看到交换机和队列的绑定关系。

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

点击Get messages标签Get message(s)按钮可以看到队列中的消息。

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

到此说明我们已经将一个消息发布到了消息队列中。完整的PHP代码如下。

<?php

/*
* 发布-订阅
* create by superrd
*/ $queueName = 'superrd';
$exchangeName = 'superrd';
$routeKey = 'superrd';
$message = 'Hello World!';
$connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd'));
$connection->connect() or die("Cannot connect to the broker!\n");
try {
$channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange(); $queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue(); $queue->bind($exchangeName, $routeKey); $exchange->publish($message,$routeKey); var_dump("[x] Sent 'Hello World!'"); } catch (AMQPConnectionException $e) {
var_dump($e);
exit();
}
$connection->disconnect();

Subscribe订阅消息

在/app路径下新建一个subscribe.php的文件

touch subscribe.php
vi subscribe.php

以下是PHP代码,和发布消息一样我们先定义好用交换机、队列、RoutingKey等变量。

$queueName = 'superrd';
$exchangeName = 'superrd';
$routeKey = 'superrd';

按照我们第二章讲到的首先建立一个连接。

$connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd'));
$connection->connect() or die("Cannot connect to the broker!\n");

新建一个信道。

$channel = new AMQPChannel($connection);

与发布消息一样新建交换机。

$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange();

新建一个队列Queue。

$queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue();

通过routeKey绑定交换机和队列。

$queue->bind($exchangeName, $routeKey);

重点来了,阻塞订阅消息。

//阻塞模式接收消息

echo "Message:\n";
while(True){
$queue->consume('processMessage');
//自动ACK应答
//$queue->consume('processMessage', AMQP_AUTOACK);
} $conn->disconnect();
/*
* 消费回调函数
* 处理消息
*/
function processMessage($envelope, $q) {
$msg = $envelope->getBody();
echo $msg."\n"; //处理消息
$q->ack($envelope->getDeliveryTag()); //手动发送ACK应答
}

注意因为是阻塞监听,因为输出缓冲区的原因用浏览器访问该文件是看不到输出的。使用脚本访问。

php /app/subscribe.php

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

通过WEB工具查看队列。superrd队列中的消息数已经为0。

(五)RabbitMQ消息队列-安装amqp扩展并订阅/发布Demo(PHP版)

完整的PHP代码如下。



<?php
/*
* 发布-订阅
* create by superrd
*/
$queueName = 'superrd';
$exchangeName = 'superrd';
$routeKey = 'superrd';
$connection = new AMQPConnection(array('host' => '10.99.121.137', 'port' => '5672', 'vhost' => '/', 'login' => 'superrd', 'password' => 'superrd'));
$connection->connect() or die("Cannot connect to the broker!\n"); $channel = new AMQPChannel($connection);
$exchange = new AMQPExchange($channel);
$exchange->setName($exchangeName);
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE);
$exchange->declareExchange(); $queue = new AMQPQueue($channel);
$queue->setName($queueName);
$queue->setFlags(AMQP_DURABLE);
$queue->declareQueue(); $queue->bind($exchangeName, $routeKey); //阻塞模式接收消息 echo "Message:\n";
while(True){
$queue->consume('processMessage');
//自动ACK应答
//$queue->consume('processMessage', AMQP_AUTOACK);
} $conn->disconnect(); /*
* 消费回调函数
* 处理消息
*/
function processMessage($envelope, $q) {
$msg = $envelope->getBody();
echo $msg."\n"; //处理消息
$q->ack($envelope->getDeliveryTag()); //手动发送ACK应答
}

RabbitMQ技术交流QQ群:327034977(添加时请备注RabbitMQ)

上一篇:PostgreSQL代码分析,查询优化部分,pull_ands()和pull_ors()


下一篇:GitHub入门:如何上传与下载工程?