平面设计网站推荐,目前比较新的网站建设技术,wordpress点击量最多的文章,网络优化工程师有多累基于php-amqplib/php-amqplib组件适配laravel框架的amqp封装库
支持便捷可配置的队列工作模式 官网详情
在此基础上可支持延迟消息、死信队列等机制。
环境要求#xff1a;
PHP版本: ^7.3|^8.0
需要开启的扩展: socket
其他:
如果需要实现延迟任务需要安装对应版本的ra…基于php-amqplib/php-amqplib组件适配laravel框架的amqp封装库
支持便捷可配置的队列工作模式 官网详情
在此基础上可支持延迟消息、死信队列等机制。
环境要求
PHP版本: ^7.3|^8.0
需要开启的扩展: socket
其他:
如果需要实现延迟任务需要安装对应版本的rabbitmq延迟插件以rabbitmq3.9.0版本为例:
wget https://github.com/rabbitmq/rabbitmq-delayed-message-exchange/releases/download/3.9.0/rabbitmq_delayed_message_exchange-3.9.0.ez
cp rabbitmq_delayed_message_exchange-3.9.0.ez /opt/rabbitmq/plugins/
rabbitmq-plugins enable rabbitmq_delayed_message_exchange用法
第一步 安装组件
composer require sai97/laravel-amqp第二步 发布服务以及配置
php artisan vendor:publish --providerSai97\LaravelAmqp\AmqpQueueProviders执行完后会分别在app/config目录下生成amqp.php(amqp连接配置等)、app/QueueJob/DefaultQueueJob.php(默认队列任务)
amqp.php
?phpuse App\QueueJob\DefaultQueueJob;return [connection [default [host env(AMQP_HOST, 127.0.0.1),port env(AMQP_PORT, 5672),user env(AMQP_USER, root),password env(AMQP_PASSWORD, root)]],event [default DefaultQueueJob::class,]
];
connection为amqp连接配置可根据自身业务去调整完全对应php-amqplib/php-amqplib相关配置项 event是队列实例标识最好和connection用相同的key以便管理。
目前可支持相关接口项 //获取连接名称public function getConnectName(): string;//获取交换机名称public function getExchangeName(): string;//获取交换机类型public function getExchangeType(): string;//获取队列名称public function getQueueName(): string;//获取路由KEYpublic function getRoutingKey(): string;//获取ContentTypepublic function getContentType(): string;//是否开启死信模式public function isDeadLetter(): bool;//获取死信交换机名称public function getDeadLetterExchangeName(): string;//获取死信路由KEYpublic function getDeadLetterRoutingKey(): string;//获取死信队列名称public function getDeadLetterQueueName(): string;//是否开启延迟任务public function isDelay(): bool;//获取延迟任务过期时长public function getDelayTTL(): int;//获取队列附加参数public function getQueueArgs(): array;//获取回调函数public function getCallback(): callable;//是否自动提交ACKpublic function isAutoAck(): bool;
当然你也可以自定义队列实例只要继承Sai97\LaravelAmqp\Queue基类即可具体功能配置参数参考Sai97\LaravelAmqp\QueueInterface。
代码示例:
生产者:
$message This is message...;
$amqpQueueServices new AmqpQueueServices(QueueFactory::getInstance(DefaultQueueJob::class));
$amqpQueueServices-producer($message);
消费者:
利用laravel自带的Command去定义一个RabbitMQWorker自定义命令行仅需要定义一次后续只需要更改amqp.php配置文件添加不同的队列实例绑定关系即可以下是RabbitMQWorker演示代码
?phpnamespace App\Console\Commands;use Illuminate\Console\Command;
use Sai97\LaravelAmqp\AmqpQueueServices;
use Sai97\LaravelAmqp\QueueFactory;class RabbitMQWorker extends Command
{/*** The name and signature of the console command.** var string*/protected $signature rabbitmq:worker {event};/*** The console command description.** var string*/protected $description rabbitmq worker 消费进程;/*** Create a new command instance.** return void*/public function __construct(){parent::__construct();}/*** Execute the console command.** return mixed*/public function handle(){try {$event $this-argument(event);$eventConfig config(amqp.event);if (!isset($eventConfig[$event]) || empty($entity $eventConfig[$event])) {return $this-error(未知的事件: {$event});}$this-info(rabbitmq worker of event[{$event}] process start ...);$amqpQueueServices new AmqpQueueServices(QueueFactory::getInstance($entity));$amqpQueueServices-consumer();} catch (\Throwable $throwable) {$event $event ?? ;$this-error($throwable-getFile() . [{$throwable-getLine()}]);return $this-error(rabbitmq worker of event[{$event}] process error:{$throwable-getMessage()});}$this-info(rabbitmq worker of event[{$event}] process stop ...);}
}
完成RabbitMQWorker消费者命令后我们只需执行php artisan rabbitmq:worker default 完成监听其中default是可变的请根据的amqp.php配置中的队列实例绑定标识去输入。
因为队列的消费者都需要是守护进程所以我们可以依托supervisord进程管理器去定义RabbitMQWorker消费者命令这样可以保证进程可后台允许以及重启启动等以下是supervisord.conf配置文件示例
[program:rabbitmq-worker-default]
#process_name%(program_name)s_%(process_num)d
process_nameworker_%(process_num)d
numprocs3
command/usr/local/bin/php /app/www/laravel8/artisan rabbitmq:worker default
autostarttrue
autorestarttrue
startretries3
priority3
stdout_logfile/var/log/rabbitmq-worker-default.log
redirect_stderrtrue搭配supervisord来进行管理消费者进程有许多便捷的方面
如果需要新增一个队列实例只需要按照上述格式复制一个program可以在不影响其他进程的情况下进程更新supervisord配置
supervisorctl update2. 通过配置numprocs参数来设定需要开启多少个相同配置项的消费者worker这在任务分发、并行处理等场景十分适用大大提高消费者执行效率。
这里不详细叙述supervisord相关操作具体可查看supervisord官方文档。 参考链接https://github.com/Z-Sai/laravel-amqp
