rabbitmq生产与消费测试
作者:互联网
消息生成者:
<?php defined('BASEPATH') OR exit('No direct script access allowed'); /* require_once FCPATH . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; */ /* set_include_path(APPPATH . '/third_party/PhpAmqpLib/Connection'); require_once APPPATH . 'third_party/PhpAmqpLib/Connection/AMQPStreamConnection.php'; require_once APPPATH . 'third_party/PhpAmqpLib/Connection/Message/AMQPMessage.php'; */ require_once FCPATH . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; $exchangeName = 'kd_sms_send_ex'; //交换机名 $queueName = 'kd_sms_send_q'; //队列名称 $routingKey = 'sms_send'; //路由关键字(也可以省略) //建立生产者与mq之间的连接 $connection = new AMQPStreamConnection('101.200.62.202', 5672, 'msgsend', 'test_123','msgs'); $channel = $connection->channel();//在已连接基础上建立生产者与mq之间的通道 $channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机 $channel->queue_declare($queueName, false, true, false, false);//声明初始化一条队列 $channel->queue_bind($queueName, $exchangeName, $routingKey); //将队列与某个交换机进行绑定,并使用路由关键字 $msgBody = json_encode(["name" => "iGoo", "age" => 22]); $msg = new AMQPMessage($msgBody, ['content_type' => 'text/plain', 'delivery_mode' => 2]); //生成消息 $channel->basic_publish($msg, $exchangeName, $routingKey);//推送消息到某个交换机 echo " [x] Sent 'Hello World!'\n"; $channel->close(); $connection->close();
消息消费者:
<?php defined('BASEPATH') OR exit('No direct script access allowed'); require_once FCPATH . '/vendor/autoload.php'; use PhpAmqpLib\Connection\AMQPStreamConnection; use PhpAmqpLib\Message\AMQPMessage; $exchangeName = 'kd_sms_send_ex'; //交换机名 $queueName = 'kd_sms_send_q'; //队列名称 $routingKey = 'sms_send'; //路由关键字(也可以省略) $connection = new AMQPStreamConnection('101.200.62.202', 5672, 'msgsend', 'test_123','msgs'); $channel = $connection->channel(); $channel->exchange_declare($exchangeName, 'direct', false, true, false); //声明初始化交换机 $channel->queue_declare($queueName, false, true, false, false); $channel->queue_bind($queueName, $exchangeName, $routingKey); //将队列与某个交换机进行绑定,并使用路由关键字 echo " [*] Waiting for messages. To exit press CTRL+C\n"; $callback = function ($msg) { echo ' [x] Received ', $msg->body, "\n"; }; $channel->basic_consume($queueName, '', false, true, false, false, $callback); while ($channel->is_consuming()) {//这个是阻塞模式,有消息就执行回调 $channel->wait(); } /* $channel->close(); $connection->close(); */
标签:消费,false,exchangeName,rabbitmq,交换机,测试,true,queueName,channel 来源: https://www.cnblogs.com/andydao/p/14783660.html