Lumen结合Horizon与RabbitMQ(完整的MQ队列方

必备条件：
1.lumen5.5+版本,低于5.5或者等于5.5会出现问题
2.已经安装好了RabbitMQ,如没有安装请翻阅的本博客的安装RabbitMQ文章
3.已经安装了composer

第一步：
安装依赖与发布资源

composer require vladimir-yuldashev/laravel-queue-rabbitmq
composer require  kinsolee/horizon-lumen
composer require  noitran/lumen-horizon-rabbitmq
composer require laravelista/lumen-vendor-publish

修改文件app/Console/Kernel.php

protected $commands = [
    \Laravelista\LumenVendorPublish\VendorPublishCommand::class
];

发布资源

php artisan vendor:publish --provider="Laravel\Horizon\HorizonServiceProvider"

第二步:
修改引导文件

$app->register(\Laravel\Horizon\HorizonServiceProvider::class);
$app->register(Noitran\Lumen\Horizon\HorizonServiceProvider::class);
$app->register(VladimirYuldashev\LaravelQueueRabbitMQ\LaravelQueueRabbitMQServiceProvider::class);

第三步
修改config/queue.php文件

'rabbitmq' => [

            'driver' => 'rabbitmq',

            'dsn' => env('RABBITMQ_DSN', null),


            'factory_class' => Enqueue\AmqpLib\AmqpConnectionFactory::class,

            'host' => env('RABBITMQ_HOST', '127.0.0.1'),
            'port' => env('RABBITMQ_PORT', 5672),

            'vhost' => env('RABBITMQ_VHOST', '/'),
            'login' => env('RABBITMQ_LOGIN', 'smj227227'),
            'password' => env('RABBITMQ_PASSWORD', '227227'),

            'queue' => env('RABBITMQ_QUEUE', 'default'),

            'options' => [

                'exchange' => [

                    'name' => env('RABBITMQ_EXCHANGE_NAME'),
                    'declare' => env('RABBITMQ_EXCHANGE_DECLARE', true),
                    'type' => env('RABBITMQ_EXCHANGE_TYPE', \Interop\Amqp\AmqpTopic::TYPE_DIRECT),
                    'passive' => env('RABBITMQ_EXCHANGE_PASSIVE', false),
                    'durable' => env('RABBITMQ_EXCHANGE_DURABLE', true),
                    'auto_delete' => env('RABBITMQ_EXCHANGE_AUTODELETE', false),
                    'arguments' => env('RABBITMQ_EXCHANGE_ARGUMENTS'),
                ],

                'queue' => [

                    'declare' => env('RABBITMQ_QUEUE_DECLARE', true),
                     'bind' => env('RABBITMQ_QUEUE_DECLARE_BIND', true),

                    /*
                     * Read more about possible values at https://www.rabbitmq.com/tutorials/amqp-concepts.html
                     */

                    'passive' => env('RABBITMQ_QUEUE_PASSIVE', false),
                    'durable' => env('RABBITMQ_QUEUE_DURABLE', true),
                    'exclusive' => env('RABBITMQ_QUEUE_EXCLUSIVE', false),
                    'auto_delete' => env('RABBITMQ_QUEUE_AUTODELETE', false),
                    'arguments' => env('RABBITMQ_QUEUE_ARGUMENTS'),
                ],
            ],

            /*
             * Determine the number of seconds to sleep if there's an error communicating with rabbitmq
             * If set to false, it'll throw an exception rather than doing the sleep for X seconds.
             */

            'sleep_on_error' => env('RABBITMQ_ERROR_SLEEP', 5),

            /*
             * Optional SSL params if an SSL connection is used
             * Using an SSL connection will also require to configure your RabbitMQ to enable SSL. More details can be founds here: https://www.rabbitmq.com/ssl.html
             */

            'ssl_params' => [
                'ssl_on' => env('RABBITMQ_SSL', false),
                '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),
            ],

        ],

第四步：
测试发布消息

<?php
/*
 *author:shumingjian
 *time 2019/3/18 5:05 PM
 *All rights reserved
*/


namespace App\Http\Controllers\Mq\RabbitMQ;


use App\Http\Controllers\Controller;
use App\Jobs\Msg;
use App\Jobs\RabbitMQJob;


class MsgController extends Controller
{
 public function test(){
     for ($i=0;$i<=1000000;$i++){
         $json = json_encode(['a'=>$i]);
         $json1 = json_encode(['a'=>'abc'.$i]);
         //\Amqp::publish('test',$json,['queue'=>'test123']);
         dispatch(new Msg($json))->onQueue('default');
         //dispatch(new RabbitMQJob($json1))->onQueue('bbb');
     }
 }
}

第六步:
作为customer，我们必须需要消费
创建Job文件

<?php
/*
    *author:shumingjian
    *time 2019/3/21 1:31 PM
    *All rights reserved
*/


namespace App\Jobs;


class Msg extends Job
{
    public $data ;

    public function __construct($data)
    {
        $this->data = $data;
    }

    public function handle(){
        $data = json_decode($this->data,true);
        print_r($data);
    }
}

第七步:
启动Horizon，执行命令

php artisan  horizon

结果应该如图所示：

WechatIMG366.jpeg