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,处理不同的消息队列。
发表评论:
◎欢迎参与讨论,请在这里发表您的看法、交流您的观点。