一、安装rabbitmq服务端
1、安装erlang
试了几次erlang官方的包安装后，在安装rabbitmq时都报错要依赖R14B02，干脆使用rabbitmq官方提供的包：https://github.com/rabbitmq/erlang-rpm 点击下载，终于不报错了。
rabbitmq官方提供安装包的步骤：
1、下载后进入安装包，直接输入命令：make
2、安装过程需要3分钟左右，编译过程中需要用到rpm-build和autoconf，没有安装的需要安装。安装完成后，会在目录RPMS/x86_64/下找到erlang的安装包，rpm -ivh *** 安装即可。
2、安装rabbitmq
我直接使用rabbitmq 的rpm安装包。如果这时还报错socat找不到，下载socat 的安装包安装即可。
注意：如果使用yum安装socat，需安装epel。
3、测试rabbitmq是否安装成功
service rabbitmq-server start #开启rabbitmq
service rabbitmq-server status #查看rabbitmq状态
二、安装rabbitmq的php扩展
1、安装扩展依赖库#####
注意:扩展是C写的,由于C与RabbitMQ通信一般需要依赖rabbitmq-c库(也就是librabbitmq),所以编译扩展前需要先装依赖库。不同版本的扩展，对php版本和librabbitmq兼容性不一样。
rabbitmq-c依赖库下载： https://github.com/alanxz/rabbitmq-c/archive/v0.8.0.tar.gz
mkdir build && cd build # 这一步是在rabbitmq-c的根目录下创建一个build子目录
cmake -DCMAKE_INSTALL_PREFIX=/usr/local/librabbitmq .. # 这一步是让cmake根据../CMakeList.txt，即rabbitmq-c的根目录下的CMakeList.txt创建Makefile文件，Makefile文件会被创建到build目录中
cmake --build . # 这一步是真正的build rabbitmq-c库的，注意，不要漏掉结尾的点 '.'
make
make install
2、安装amqp扩展
amqp扩展下载： http://pecl.php.net/get/amqp-1.9.0.tgz
tar zvxf amqp-1.9.0.tgz #解压
cd amqp-1.9.9 #打开目录
./configure --with-php-config=/usr/local/php56/bin/php-config --with-amqp --with-librabbitmq-dir=/usr/local/librabbitmq
make
make install
修改php.ini
extension=amqp.so #extension_dir自己定义
重启php-fpm
service php-fpm restart #重启
如无意外，则可以在phpinfo页面看到如下所示：
三、监控及代码测试
启动rabbitmq监控页面
rabbitmq-plugins enable rabbitmq_management #启动监控页面插件
生产者代码 (创建一个rabbit_publisher.php的文件)
创建连接-->创建channel-->创建交换机对象-->发送消息
<?php
//配置信息
$conn_args = array(
'host' => '192.168.1.93',
'port' => '5672',
'login' => 'guest',
'password' => 'guest',
'vhost'=>'/'
);
$e_name = 'e_linvo'; //交换机名
//$q_name = 'q_linvo'; //无需队列名
$k_route = 'key_1'; //路由key
//创建连接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//消息内容
$message = "TEST MESSAGE! 测试消息！";
//创建交换机对象
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
//发送消息
//$channel->startTransaction(); //开始事务
for($i=0; $i<5; ++$i){
echo "Send Message:".$ex->publish($message, $k_route)."\n";
}
//$channel->commitTransaction(); //提交事务
$conn->disconnect();
消费者代码(创建一个rabbit_consumer.php的文件)
创建连接-->创建channel-->创建交换机-->创建队列-->绑定交换机/队列/路由键-->接收消息
<?php
//配置信息
$conn_args = array(
'host' => '192.168.1.93',
'port' => '5672',
'login' => 'guest',
'password' => 'guest',
'vhost'=>'/'
);
$e_name = 'e_linvo'; //交换机名
$q_name = 'q_linvo'; //队列名
$k_route = 'key_1'; //路由key
//创建连接和channel
$conn = new AMQPConnection($conn_args);
if (!$conn->connect()) {
die("Cannot connect to the broker!\n");
}
$channel = new AMQPChannel($conn);
//创建交换机
$ex = new AMQPExchange($channel);
$ex->setName($e_name);
$ex->setType(AMQP_EX_TYPE_DIRECT); //direct类型
$ex->setFlags(AMQP_DURABLE); //持久化
echo "Exchange Status:".$ex->declare()."\n";
//创建队列
$q = new AMQPQueue($channel);
$q->setName($q_name);
$q->setFlags(AMQP_DURABLE); //持久化
echo "Message Total:".$q->declare()."\n";
//绑定交换机与队列，并指定路由键
echo 'Queue Bind: '.$q->bind($e_name, $k_route)."\n";
//阻塞模式接收消息
echo "Message:\n";
while(True){
$q->consume('processMessage');
//$q->consume('processMessage', AMQP_AUTOACK); //自动ACK应答
}
$conn->disconnect();
/**
* 消费回调函数
* 处理消息
*/
function processMessage($envelope, $queue) {
$msg = $envelope->getBody();
echo $msg."\n"; //处理消息
$queue->ack($envelope->getDeliveryTag()); //手动发送ACK应答
}
/**
* 消费回调函数
* 处理消息
*/
function processMessage($envelope, $queue) {
var_dump($envelope->getRoutingKey);
$msg = $envelope->getBody();
echo $msg."\n"; //处理消息
}
需要注意的地方是：
queue对象有两个方法可用于取消息：consume和get。前者是阻塞的，无消息时会被挂起，适合循环中使用；
后者则是非阻塞的，取消息时有则取，无则返回false。
测试截图
运行消费者：
运行生产者，发消息：
消费者接收到消息：
执行两个文件，再打开RabbitMQ的管理中心 http://127.0.0.1:15672/
