laravel8-rabbitmq消息队列-实时监听跨服务器消息

使用场景介绍：
1）用于实时监听远程服务器发出的消息（json格式消息），接受并更新消息状态，存储到本地服务器
2）环境：lNMP（laravel8）
3）服务器需要开启rabbitmq驱动队列

1、composer安装rabbitmq扩展包

vladimir-yuldashev/laravel-queue-rabbitmq
参考文档：[https://blog.csdn.net/u012321434/article/details/126246141]

2、安装配置文件

1. 打开app/config/queue.php中connections数组中添加以下代码，根据实际情况填写相关配置信息

 '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' => ['verify_peer' => env('RABBITMQ_SSL_VERIFY_PEER', false),],'queue' => [//此处直接添加到自定义的job任务中'job' => App\Jobs\Rabbitmq\RabbitMQJob::class,//以下配置是rabbitmq 广播模式(direct)'exchange' => 'amq','exchange_type' => 'direct','exchange_routing_key' => '',],],/** Set to "horizon" if you wish to use Laravel Horizon.*/'worker' => env('RABBITMQ_WORKER', 'default'),],

2. .env文件中配置相关参数信息

RABBITMQ_HOST=127.0.0.1
RABBITMQ_PORT=5672
RABBITMQ_USER=testuser
RABBITMQ_PASSWORD=test
RABBITMQ_VHOST=/project
RABBITMQ_QUEUE=que_project

4. 在app/config/logging.php文件channels选项中添加自定义log日志，记录报错日志信息

 'rabbitmq' => ['driver' => 'daily','path' => storage_path('logs/rabbitmq.log'),'level' => env('LOG_LEVEL', 'debug'),'days' => 14,],

4. RabbitMQJob.php

namespace App\Jobs\Rabbitmq;use Illuminate\Support\Str;
use VladimirYuldashev\LaravelQueueRabbitMQ\Queue\Jobs\RabbitMQJob as BaseJob;
use App\Services\Rabbitmq\RabbitmqService;class RabbitMQJob extends BaseJob
{public $tries = 1;public $timeout = 3600;public $maxExceptions = 3;public function fire(){$payload = $this->payload();(new RabbitmqService())->handle($payload['data']);$this->delete();}/*** Get the decoded body of the job.* 接收消息体并自定义处理* @return array*/public function payload(){return ['uuid' => (string) Str::uuid(),'job'  => '\App\Services\Rabbitmq\RabbitmqService@handle','maxTries' => $this->tries,'maxExceptions' => $this->maxExceptions,'timeout' => $this->timeout,'data' => json_decode($this->getRawBody(), true)];}/*** Process an exception that caused the job to fail.** @param  \Throwable|null  $e* @return void*/protected function failed($e){(new RabbitmqService())->failed($e);}}

6. RabbitmqService.php

namespace App\Services\Rabbitmq;use Illuminate\Support\Facades\Log;class RabbitmqService
{protected $logName = 'rabbitmq';protected $connection;protected $channel;public $messageService;/*** 处理消息状态* @param $message .接收到的消息* @return bool*/public function handle($message=''){//1.判断接收的消息情况Log::channel($this->logName)->info('接收的消息体：'.json_encode($message));//接收到的消息$message = json_decode(json_encode($message), true);//2.消息自定义处理}/*** 异常扑获* @param \Exception $exception*/public function failed(\Exception $exception){Log::channel($this->logName)->info('异常：'.json_encode($exception->getMessage()));}}

8. 服务器开启rabbitmq队列驱动，开始监听消息

php artisan queue:work rabbitmq