thinkphp5.1 之RabbitMq应用实践
参考原作者文章：《thinkphp5.1应用rabbitMq》
https://blog.csdn.net/qq_42137846/article/details/125842245
为了验证实践过程，我做了相应的改动。
RabbitMQ的安装请参考《centos7.5 安装RabbitMQ Erlang》
△配置Rabbit文件
路径如下：
application|extra|rabbitmq.php
代码如下
<?php
// +----------------------------------------------------------------------
// | rabbitMQ 示例配置
//
// Exchange在定义的时候是有类型的，以决定到底是哪些Queue符合条件，可以接收消息：
// fanout：所有bind到此exchange的queue都可以接收消息
// direct：通过routingKey和exchange决定的那个唯一的queue可以接收消息
// topic：所有符合routingKey(此时可以是一个表达式)的routingKey所bind的queue可以接收消息
// headers：通过headers 来决定把消息发给哪些queue(这个很少用)
// +----------------------------------------------------------------------
return [
// 连接信息
'AMQP' => [
'host' => '192.168.66.64',
'port' => '5672',
'username' => 'admin',
'password' => 'admin',
'vhost' => '/'
],
//测试队列：direct 路由模式
'test_queue' => [
'exchange_name' => 'test_exchange', //交换机名称
'exchange_type' => 'direct', //交换机运行模式(从上面四种模式中选)
'queue_name' => 'test_queue', //队列名称
'route_key' => 'test', //路由键,用于绑定队列与交换机
'consumer_tag' => 'test' //消费标签
],
//订单队列：direct 路由模式
'order_queue' => [
'exchange_name' => 'order_exchange', //交换机名称
'exchange_type' => 'direct', //交换机运行模式(从上面四种模式中选)
'queue_name' => 'order_queue', //队列名称
'route_key' => 'order', //路由键,用于绑定队列与交换机
'consumer_tag' => 'order' //消费标签
],
//...等等其它队列
];
△生产者
路径如下：
application|common|controller|MqProducer.php
代码如下：
<?php
namespace app\common\controller;
use think\Config;
//引入rabbitMq所需类
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Message\AMQPMessage;
class MqProducer
{
/**
* 生产者
* @param $data 控制器传来的数据
* @param $name 配置文件下rabbitmq文件前缀名
* @throws \Exception
*/
public function publish($data, $name)
{
//获取rabbitMq配置(刚刚在config文件下配置的rabbitmq信息)
$amqp = Config::get('rabbitmq.AMQP');
$amqpDefail = Config::get('rabbitmq.' . $name . '_queue');
//建立连接
$connection = new AMQPStreamConnection(
$amqp['host'],
$amqp['port'],
$amqp['username'],
$amqp['password']
);
//建立通道
$channel = $connection->channel();
//初始化交换机
$channel->exchange_declare($amqpDefail['exchange_name'], $amqpDefail['exchange_type'], false, true, false);
//初始化队列
$channel->queue_declare($amqpDefail['queue_name'], false, true, false, false);
//绑定队列到交换机
$channel->queue_bind($amqpDefail['queue_name'], $amqpDefail['exchange_name'], $amqpDefail['route_key']);
//生成消息(json格式传输)
$msg = new AMQPMessage(json_encode($data, JSON_UNESCAPED_UNICODE | JSON_UNESCAPED_SLASHES));
//推送消息到交换机
$channel->basic_publish($msg, $amqpDefail['exchange_name'], $amqpDefail['route_key']);
//关闭通道
$channel->close();
//断开连接
$connection->close();
}
}
△消费者
文件路径：
application|common|controller|MqConsumer.php
代码如下：
<?php
namespace app\common\controller;
use think\Config;
use think\Log;
//引入rabbitMq所需类
use PhpAmqpLib\Message\AMQPMessage;
use PhpAmqpLib\Connection\AMQPStreamConnection;
use PhpAmqpLib\Exchange\AMQPExchangeType;
class MqConsumer
{
/**
* 消费者(路由模式)
* @param $name 配置文件下rabbitmq文件前缀名
* @throws \ErrorException
*/
public function consumer($name)
{
//获取rabbitMq配置
$amqp = Config::get('rabbitmq.AMQP');
$amqpDefail = Config::get('rabbitmq.' . $name . '_queue');
//建立连接
$connection = new AMQPStreamConnection(
$amqp['host'],
$amqp['port'],
$amqp['username'],
$amqp['password']
);
//建立通道
$channel = $connection->channel();
//流量控制(也被称为公平调度，这里设置一个消费者一次只处理200条消息)
$channel->basic_qos(null, 200, null);
//初始化交换机
$channel->exchange_declare($amqpDefail['exchange_name'], $amqpDefail['exchange_type'], false, true, false);
//初始化队列
$channel->queue_declare($amqpDefail['queue_name'], false, true, false, false);
//绑定队列与交换机
$channel->queue_bind($amqpDefail['queue_name'], $amqpDefail['exchange_name'], $amqpDefail['route_key']);
//消费消息
$channel->basic_consume($amqpDefail['queue_name'], $amqpDefail['consumer_tag'], false, false, false, false, [$this, 'callback']);
//注册退出函数行为(其实用不到,因为要保证消费者一直运行,所以不能断开连接))
// register_shutdown_function([$this,'shutdown'], $channel, $connection);
//消息未处理完毕时,循环监听并一直处理上方callback方法的逻辑
$channel->consume();
}
/**
* 回调后消息处理(业务逻辑放置此处)
* @param $msg
*/
public function callback($msg)
{
//$msg->body 通过消费者传来的控制器接收的数据
$data = json_decode($msg->body,true);//拿到数据
Log::write('拿到的消费者数据为:'.json_encode($data,320),'info');
//这里我拿来做了一个插库的动作
/*app('weloginper')->create([
'ToUserName' => 'test',
'FromUserName' => 'test',
'CreateTime' => 'test',
'MsgType' => 'test',
'Event' => 'test',
'EventKey' => 'test',
'phone' => 'test',
]);*/
//如果有个值是quit,则让该消费者停止消费
/*if($data['aa'] == 'quit'){
$msg->getChannel()->basic_cancel($msg->getConsumerTag());
}*/
$msg->ack(); //消息应答：这波200条消息处理完毕后进行消息确认,告诉mq可以开始发下一波200条消息了
}
/**
* @param $channel 信道
* @param \PhpAmqpLib\Connection\AbstractConnection $connection
*/
function shutdown($channel, $connection)
{
$channel->close(); //关闭通道
$connection->close(); //断开连接
}
}
生产者业务动作
目的：模拟生成消息
文件路径：application|common|controller|Index.php
代码如下：
<?php
namespace app\common\controller;
use think\Console;
use app\common\controller\MqProducer;
class Index extends MqProducer
{
public function index()
{
$page = input('get.page') ? input('get.page') : 1;//得到初始页码
$data = [
'page' => $page, //当前页
];
if($page <= 100){ //当前不是第10页则一直调用刚刚定义的生产者publish去生产消息
self::publish($data, 'order');
$url = url('/common/index/index') . '?page=' . ++$page;
die("<script>location.href='{$url}';</script>");
}
return json([
'code' => 200,
'msg' => 'order队列推送成功'
]);
}
}
配置下config
config文件默认禁止访问common模块了
// 禁止访问模块 原来的
//'deny_module_list' => ['common', 'admin'],
//禁止访问模块 修改后的
'deny_module_list' => ['admin'],
生成消息
访问 xx.com/common/index/index
查看RabbitMQ
消费者消费
命令行：think php order
再查看RabbitMq 发现消息被消费了。
