laravel之rabbitmq组件使用

以你之姓@ 2022-10-06 15:57 265阅读 0赞

# 介绍 #

laravel入门及技术指南，参见相关文章。laravel集成使用消息中间件rabbitmq，提供服务器异步消息处理，提升服务器性能。下面谈谈如何安装及集成使用。

# 安装 #

安装环境为centos7.6，php7.2.33

## 安装rabbitmq ##

//拉取镜像
    # docker pull rabbitmq
    //生成容器
    # docker run -p 5672:5672 -p 15672:15672 -d --name rabbitmq -v /docker/rabbitmq:/var/lib/rabbitmq --privileged=true -e RABBITMQ_DEFAULT_VHOST=my_vhost -e RABBITMQ_DEFAULT_USER=admin -e RABBITMQ_DEFAULT_PASS=admin rabbitmq
    //查看容器状态
    # docker ps
    //进入容器
    # docker exec -it rabbitmq bash
    进入容器内安装rabbitmq_management
    rabbitmq-plugins enable rabbitmq_management

成功后，浏览器访问http://ip:15672/   登录 admin/admin

![20210615174736965.png][]![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70][]

## 安装php的rabbitmq扩展 ##

**查找、安装rabbitmq包**。

\# yum search rabbitmq

\# yum install -y librabbitmq.x86\_64

**安装php的amqp扩展**

**找到phpize**，php编译工具

xxx/php/72/bin/**phpize**

**准备素材包**amqp-1.10.2.tgz、rabbitmq-c-0.8.0.tar.gz，放到/home下

//解压缩
    home# tar zxvf amqp-1.10.2.tgz
    home#  tar zxvf rabbitmq-c-0.8.0.tar.gz
    //生成编译文件 configure
    home/amqp-1.10.2# xxx/php/72/bin/phpize
    home/rabbitmq-c-0.8.0#  xxx/php/72/bin/phpize
    //这样amqp-1.10.2、rabbitmq-c-0.8.0下都生成了configure编译文件
    编译安装
    /home/rabbitmq-c-0.8.0#./configure --prefix=/home/rabbitmq-c-0.8.0
    /home/rabbitmq-c-0.8.0# make && make install
    
    home/amqp-1.10.2# ./configure --with-php-config=/www/server/php/72/bin/php-config --with-amqp --with-librabbitmq-dir=/home/rabbitmq-c-0.8.0
    home/amqp-1.10.2# make && make install
    成功后生成amqp.so，自动放在php扩展目录下，然后对相应php.ini中配置如下：
    [rabbitmq]
    extension = xxx/php/72/lib/php/extensions/no-debug-non-zts-20170718/amqp.so

我们查看php -m，看是否安装成功

![20210615173420717.png][]

# 代码集成 #

# laravel框架提供了rabbitmq集成 #

laravel需要composer安装的依赖包如下，采用laravels（laravel+swoole）提高性能。
    "dingo/api": "^3.0",
    "hhxsv5/laravel-s": "^3.7",
    "vladimir-yuldashev/laravel-queue-rabbitmq": "10.X"

### **laravel配置rabbitmq** ###

.env文件

...
    #laravel默认连接为sync，改为rabbitmq 
    QUEUE_CONNECTION=rabbitmq
    
    #增加rabbitmq 
    QUEUE_DRIVER=rabbitmq
    RABBITMQ_HOST=172.17.0.3
    RABBITMQ_PORT=5672
    RABBITMQ_USER=admin
    RABBITMQ_PASSWORD=admin
    RABBITMQ_VHOST=my_vhost
    RABBITMQ_QUEUE=lmrs

注： 上述172.17.0.3是rabbitmq安装后的容器ip。

config/queue.php，修改增加rabbitmq配置

'rabbitmq' => [
                'driver' => 'rabbitmq',
                'queue' => env('RABBITMQ_QUEUE', 'default'),
                'connection' => PhpAmqpLib\Connection\AMQPLazyConnection::class,
    
                'hosts' => [
                    [
                        'host' => env('RABBITMQ_HOST', '127.0.0.1'),
                        'port' => env('RABBITMQ_PORT', 5672),
                        'user' => env('RABBITMQ_USER', 'guest'),
                        'password' => env('RABBITMQ_PASSWORD', 'guest'),
                        'vhost' => env('RABBITMQ_VHOST', '/'),
                    ],
                ],
    
                'options' => [
                    'ssl_options' => [
                        'cafile' => env('RABBITMQ_SSL_CAFILE', null),
                        'local_cert' => env('RABBITMQ_SSL_LOCALCERT', null),
                        'local_key' => env('RABBITMQ_SSL_LOCALKEY', null),
                        'verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', true),
                        'passphrase' => env('RABBITMQ_SSL_PASSPHRASE', null),
                    ],
                    'queue' => [
                        'job' => \VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob::class,
                    ]
                ],
    
                /*
                 * Set to "horizon" if you wish to use Laravel Horizon.
                 */
                'worker' => env('RABBITMQ_WORKER', 'default'),
            ]

###  业务代码用例 ###

服务类RabbitmqService：主要提供了rabbitmq服务的连接、推送到消息队列、取出消息等，这些基本方法。  
services/RabbitmqService.php

class RabbitmqService{
        //取得rabbitmq连接
        public static function getConnect()    {
            $config = [
                'host' => env('RABBITMQ_HOST', '127.0.0.1'),
                'port' => env('RABBITMQ_PORT', 5672),
                'user' => env('RABBITMQ_USER', 'guest'),
                'password' => env('RABBITMQ_PASSWORD', 'guest'),
                'vhost' => env('RABBITMQ_VHOST', '/'),
            ];
    
            return new AMQPStreamConnection($config["host"],$config["port"],$config["user"],$config["password"],$config["vhost"]);
        }
    //推送到消息队列
        public static function push($queue,$messageBody,$exchange='router')    {
            $connection = self::getConnect();
            $channel = $connection->channel();
            //声明一个队列
            $channel->queue_declare($queue,false,true,false,false);
            $channel->exchange_declare($exchange,'direct',false,true,false);
            $channel->queue_bind($queue,$exchange);
            $message = new  AMQPMessage($messageBody,array('content_type' => 'text/plain', 'delivery_mode' => AMQPMessage::DELIVERY_MODE_PERSISTENT));
            $channel->basic_publish($message,$exchange);
            $channel->close();
            $connection->close();
        }
    //取出消息
        public static function pop($queue,$callback,$exchange='router')
        {
            $connection = self::getConnect();
            $channel = $connection->channel();
            $message = $channel->basic_get($queue);
            $res = $callback($message->getBody());
            if ($res){
                $channel->basic_ack($message->getDeliveryTag());
            }
            $channel->close();
            $connection->close();
        }
    }

**消息处理类**：主要将消息通过调用上述消息服务连接、推送、取出基本接口调用，进行消息的处理。

Jobs/UpdateProduct.php，本例给出一个获取消息，推送消息队列，并在redis缓存消息表示消费消息，rabbitmq的内部ack应答机制保证消息可靠，保证在执行消息读取后处理（这是只是redis缓存当作处理）能完整完成，不会因故障导致消息丢失。

class UpdateProduct implements ShouldQueue{
        use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
    
        protected  $productKey;
    
        /**
         * Create a new job instance.
         * 构造方法传入数据消息，生成消息键值，并放入消息队列
         * @return void
         */
        public function __construct($data)
        {
            $this->productKey = "Lmrs::product::info::".$data->id;
    
            RabbitmqService::push('update_queue',$data);
        }
    
        /**
         * Execute the job.
         * 取出消息的处理
         * @return void
         */
        public function handle()
        {
            RabbitmqService::pop('update_queue',function ($message){
                $product = app('redis')->set($this->productKey,serialize($message));
                if (!$product){
                    return;
                }
                print_r($message);
            });
        }
    //异常或失败的处理
        public function failed(\Exception $exception)   {
            print_r($exception->getMessage());
        }
    }

**业务产生消息的控制类**：业务中产生消息，并调用上述消息处理器，进入rabbitmq服务器。

ProductController.php。本例模拟一个更新商品名称信息的简单用例，将更新消息推送给rabbitmq进行处理。

class ProductController extends Controller{
    ...
        //测试消息中间件rabbitmq
        public function update(Request $request)   {
            $product = Product::find($request->input("id"));
            $product->update([
                "name" => $request->input("name"),
                "long_name" => $request->input("long_name")
            ]);
            //return Product::find($request->input("id"));
            if ($product){
                $this->dispatch(new UpdateProduct(Product::find($request->input("id"))));
                return "success";
            }
            return "Failed";
        }
    }

# 运行测试 #

## 测试 ##

我们用postman工具，模拟发送请求

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70 1][]

## 消息监听 ##

laravel框架提供rabbitmq监听工具，上述配置好laravel的rabbitmq相关参数，运行下面

# php artisan queue:work rabbitmq

注：有时会报错“pcntl\_signal() has been disabled for security reasons”等，需要取消php的禁用函数pcntl\_signal、pcntl\_alarm

运行正常后，接到发来的消息，显示如下：

![20210615170813699.png][]

### rabbitmq服务器查看消息 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70 2][]

[20210615174736965.png]: https://img-blog.csdnimg.cn/20210615174736965.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70]: /images/20221005/fb9ab92d318b4c9bac41febd82aa5dfa.png
[20210615173420717.png]: /images/20221005/b16d7c4d852f4630b733c248b32f389f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70 1]: /images/20221005/ed6c8e63cb974184824edb578ac29f40.png
[20210615170813699.png]: /images/20221005/8d88746d46ea404fb278a342ca3c3356.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lhbl9kaw_size_16_color_FFFFFF_t_70 2]: /images/20221005/091fb6f0fd4043919ca9e8b2d2d5001a.png