php操作rabbitmq教程

1: 连接rabbitmq 新建exchange和queue

amqp_manager.php

$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest');
$conn = new AMQPConnection($conn_args);
if ($conn->connect()) {
echo "Established a connection to the broker \n";
}
else {
echo "Cannot connect to the broker \n ";
exit(0);
}

$channel = new AMQPChannel($conn);

$exchange = new AMQPExchange($channel);
$exchange->setName('lizhifeng');
$exchange->setType(AMQP_EX_TYPE_DIRECT);
$exchange->setFlags(AMQP_DURABLE | AMQP_AUTODELETE) ;
$exchange->declare(); // 声明一个名为 lizhifeng的 路由器

// 添加一个名为queue1 的队列并绑定 key1
$queue = new AMQPQueue($channel);
$queue->setName('queue1');
$queue->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);
$queue->declare();
$queue->bind('lizhifeng','key1');


// 添加一个名为queue2 的队列并绑定 key2
$queue = new AMQPQueue($channel);
$queue->setName('queue2');
$queue->setFlags(AMQP_DURABLE | AMQP_AUTODELETE);
$queue->declare();
$queue->bind('lizhifeng','key2');

// 将queue1 绑定到key3 注意key3不会覆盖key1
// 而是key1和key3将同时生效
$queue = new AMQPQueue($channel);
$queue->setName('queue1');
##$queue->setFlags(AMQP_DURABLE | AMQP_AUTODELETE); 无需重复设置队列queue1的属性
##$queue->declare(); 这里不需要再重复申明了
$queue->bind('lizhifeng','key3');


/*
// 删除exchange
$exchange = new AMQPExchange($channel);
$exchange->setName('lizhifeng');
$exchange->delete();

// 删除队列
$queue = new AMQPQueue($channel);
$queue->setName('queue1');
$queue->delete();
$queue = new AMQPQueue($channel);
$queue->setName('queue2');
$queue->delete();
*/
?>

2:连接rabbitmq 往exchange中写消息

amqp_server.php

$routingkey='key1';

$conn_args = array('host' => 'localhost', 'port' => '5672', 'login' => 'guest', 'password' => 'guest');
$conn = new AMQPConnection($conn_args);
if ($conn->connect()) {
echo "Established a connection to the broker \n";
}
else {
echo "Cannot connect to the broker \n ";
}

$channel = new AMQPChannel($conn);
$exchange = new AMQPExchange($channel);
$exchange->setName('lizhifeng');

for($i=0;$i{
if($routingkey=='key1')
{
$routingkey='key2'; // 路由到队列queue2
}
else if($routingkey=='key2')
{
$routingkey='key3'; // 路由到队列queue1
}
else
{
$routingkey='key1'; // 路由到队列queue1
}
$tmp=array();
$tmp[]="第".$i."个消息的key为".$routingkey ;
$message = json_encode($tmp);
if($exchange->publish($message,$routingkey))
{
print $routingkey."\tok\n";
}
else
{
print "error\n" ;
}
}

3:连接rabbitmq消费消息

amqp_client.php

//连接RabbitMQ
$conn_args = array( 'host'=>'127.0.0.1' , 'port'=> '5672', 'login'=>'guest' , 'password'=> 'guest','vhost' =>'/');
$conn = new AMQPConnection($conn_args);

if ($conn->connect()) {
echo "Established a connection to the broker \n";
}
else {
echo "Cannot connect to the broker \n ";
exit();
}


$channel = new AMQPChannel($conn);
$q = new AMQPQueue($channel);
$q->setName('queue1');
// 这里并不是创建新的队列,只是连接到名为quene1的队列
// 我的理解为队列其实在服务器上,消息已经被路由到不同的队列了, 我们只需取消息
while($messages = $q->get(AMQP_AUTOACK))
{
var_dump(json_decode($messages->getBody(), true ));
}

$q = new AMQPQueue($channel);
$q->setName('queue2');
while($messages = $q->get(AMQP_AUTOACK))
{
var_dump(json_decode($messages->getBody(), true ));
}

$conn->disconnect();
?>