参考原作者文章：《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(" "); } return json([ 'code' => 200, 'msg' => 'order队列推送成功' ]); } } 配置下configconfig文件默认禁止访问common模块了// 禁止访问模块 原来的 //'deny_module_list' => ['common', 'admin'], //禁止访问模块 修改后的 'deny_module_list' => ['admin'], 生成消息访问 xx.com/common/index/index查看RabbitMQ消费者消费命令行：think php order再查看RabbitMq 发现消息被消费了。
场景：category_ids 值为1,2,4查询 category_ids = 2 和 4的所有记录查询语句：$map['category_id']=array('in',$category_ids)❌$category_ids =1,2,3 $map[] =['category_id','in',$category_ids]; ✔