RabbitMQ:简单的消息中间件使用总结
1.消息发布步骤：
建立连接：$connection = new AMQPConnection(array('127.0.0.1','port'=>'5672','vhost'=>'/','login'=>'用户名','password'=>'密码'));
$connection->connect() or die("连接失败 \r\n");
建立信道：$channel = new AMQPChannel();
建立交换机：$exchange = new AMQPExchange($channel);
$exchange->setName('交换机名称');
$exchange->setType('交换机类型'); （direct（AMQP_EX_TYPE_DIRECT）、topic(AMQP_EX_TYPE_TOPIC) 和 fanout(AMQP_EX_TYPE_FANOUT)）
$exchange->setFlag(AMQP_DURABLE);
$exchange->declareExchnge();
新建队列： $queue = new AMQPQueue($channel);
$queue->setName('队列名称');
$queue->setFlag(AMQP_DURABLE);
$queue->declareQueue();
通过route_key绑定交换机和队列：
$queue->bind('交换机名称'，‘routeKey’);
发送消息： $exchange->publish("消息内容"，'routeKey');
消息持久化发布：$exchange->publish("消息内容", 'routeKey', AMQP_NOPARAM, array('delivery_mode'=>2));
断开连接： $connection->disconnect();
2、消息订阅
建立连接 $connection = new AMQPConnection(array('127.0.0.1','port'=>'5672','vhost'=>'/','login'=>'用户名','password'=>'密码'));
$connection->connect() or die("连接失败 \r\n");
新建信道： $channel = new AMQPChannel();
新建交换机： $exchange = new AMQPExchange($channel);
$exchange->setName(‘交换机名称’);
$exchange->setType('交换机类型')； （direct（AMQP_EX_TYPE_DIRECT）、topic(AMQP_EX_TYPE_TOPIC) 和 fanout(AMQP_EX_TYPE_FANOUT)）
$exchange->setFlag(AMQP_DURABLE);
$exchange->declareExchange();
新建队列： $queue = new AMQPQueue($channel);
$queue->setName('队列名称')；
$queue->setFlag(AMQP_DURABLE);
$queue->declareQueue();
通过routeKey 绑定交换机和队列
$queue->bind('交换机名称', routeKey);
订阅消息（阻塞）
while(true){
$queue->consume('perocess');
}
function perocess($event, $q){
$msg= $event->getBody();
echo $msg;
$q->ack($event->getDeliveryTag()); //手动发送ACK答应
}
交换机fanout（AMQP_EX_TYPE_FANOUT）方式使用说明：fanout模式模式不像direct模式通过routingkey来进行匹配，而是会把消息发送到所以的已经绑定的队列中。
fanout模式中routeKey为空，不需要命名，订阅对象绑定后直接进行消费，不同的消费队列消费的是相同的数据。
交换机topic（AMQP_EX_TYPE_TOPIC）方式使用说明：通过设置不同的routeKey，队列绑定不同的routeKey,通过订阅不同的routeKey,处理不同的消息队列。
网友评论