标签 thinkphp5.1 下的文章

参考原作者文章:《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
生成消息.png

查看RabbitMQ

4.png

消费者消费

命令行:think php order
10.png

再查看RabbitMq 发现消息被消费了。
2501.png

场景: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];