当前位置: 首页 > news >正文

laravel实现AMQP(rabbitmq)生产者以及消费者

基于php-amqplib/php-amqplib组件适配laravel框架的amqp封装库

支持便捷可配置的队列工作模式 官网详情

在此基础上可支持延迟消息、死信队列等机制。

环境要求:

PHP版本: ^7.3|^8.0

需要开启的扩展: socket

其他:

  1. 如果需要实现延迟任务需要安装对应版本的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 --provider="Sai97\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_name=worker_%(process_num)d
numprocs=3
command=/usr/local/bin/php /app/www/laravel8/artisan rabbitmq:worker default
autostart=true
autorestart=true
startretries=3
priority=3
stdout_logfile=/var/log/rabbitmq-worker-default.log
redirect_stderr=true

搭配supervisord来进行管理消费者进程有许多便捷的方面:

  1. 如果需要新增一个队列实例,只需要按照上述格式复制一个program,可以在不影响其他进程的情况下进程更新supervisord配置:
supervisorctl update

    2. 通过配置numprocs参数来设定需要开启多少个相同配置项的消费者worker,这在任务分发、并行处理等场景十分适用,大大提高消费者执行效率。

这里不详细叙述supervisord相关操作,具体可查看supervisord官方文档。

参考链接:https://github.com/Z-Sai/laravel-amqp

相关文章:

laravel实现AMQP(rabbitmq)生产者以及消费者

基于php-amqplib/php-amqplib组件适配laravel框架的amqp封装库 支持便捷可配置的队列工作模式 官网详情 在此基础上可支持延迟消息、死信队列等机制。 环境要求&#xff1a; PHP版本: ^7.3|^8.0 需要开启的扩展: socket 其他: 如果需要实现延迟任务需要安装对应版本的ra…...

LeetCode——二叉树篇(九)

刷题顺序及思路来源于代码随想录&#xff0c;网站地址&#xff1a;https://programmercarl.com 目录 669. 修剪二叉搜索树 108. 将有序数组转换为二叉搜索树 538. 把二叉搜索树转换为累加树 669. 修剪二叉搜索树 给你二叉搜索树的根节点 root &#xff0c;同时给定最小边界…...

uniapp scroll-view横向滚动无效,scroll-view子元素flex布局不生效

要素排查&#xff1a; 1.scroll-x属性需要开启&#xff0c;官方类型是Boolean&#xff0c;实际字符串也行。 2scroll-view标签需要给予一个固定宽度&#xff0c;可以是百分百也可以是固定宽度或者100vw。 3.子元素需要设置display: inline-block&#xff08;行内块元素&#x…...

无涯教程-进程 - 简介

进程间通信就是在不同进程之间传播或交换信息&#xff0c;那么不同进程之间存在着什么双方都可以访问的介质呢?进程的用户空间是互相独立的&#xff0c;一般而言是不能互相访问的&#xff0c;唯一的例外是共享内存区。另外&#xff0c;系统空间是“公共场所”&#xff0c;各进…...

HTML番外篇(四)-HTML5新增元素-CSS常见函数-理解浏览器前缀-BFC

一、HTML5新增元素 1.HTML5语义化元素 在HMTL5之前&#xff0c;我们的网站分布层级通常包括哪些部分呢&#xff1f; header、nav、main、footer ◼ 但是这样做有一个弊端&#xff1a; 我们往往过多的使用div, 通过id或class来区分元素&#xff1b;对于浏览器来说这些元素不…...

机器学习之Adam(Adaptive Moment Estimation)自适应学习率

Adam&#xff08;Adaptive Moment Estimation&#xff09;是一种常用的优化算法&#xff0c;特别适用于训练神经网络和深度学习模型。它是一种自适应学习率的优化算法&#xff0c;可以根据不同参数的梯度信息来动态调整学习率&#xff0c;以提高训练的效率和稳定性。 Adam算法…...

深入理解Linux权限管理:保护系统安全的重要措施

Linux操作系统以其稳定性、可靠性和灵活性而受到广泛使用。其中一个关键特性是其强大的权限管理系统&#xff0c;它可以保护系统资源和用户数据的安全性。本文将深入探讨Linux权限管理的概念、原则和实践&#xff0c;帮助您理解如何正确配置和管理权限&#xff0c;以确保系统的…...

kafka复习:(20):消费者拦截器的使用

一、定义消费者拦截器&#xff08;只消费含"sister"的消息&#xff09; package com.cisdi.dsp.modules.metaAnalysis.rest;import org.apache.kafka.clients.consumer.ConsumerInterceptor; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.…...

水库大坝安全监测的主要内容包括哪些?

在水库大坝的实时监测中&#xff0c;主要任务是通过无线传感网络监测各个监测点的水位、水压、渗流、流量、扬压力等数据&#xff0c;并在计算机上用数据模式或图形模式进行实时反映&#xff0c;以掌握整个水库大坝的各项变化情况。大坝安全监测系统能实现全天候远程自动监测&a…...

Cadence软件屏幕显示问题

问题 就是今天打开Cadence软件想导出网表看一下&#xff0c;发现没有显示确定按钮什么的&#xff0c;那个窗口也是无语&#xff0c;不能移动&#xff0c;缩放也只能左右缩放&#xff0c;还不能缩小什么的&#xff0c;真的醉了&#xff0c;后面就是调整窗口的分辨率。 因为我最…...

访问服务器快慢的因素

我们在租用服务器的过程中&#xff0c;可能在访问速度方面&#xff0c;会受到某些因素影响&#xff0c;如果您要进行此项业务&#xff0c;进行一些简单的了解 是非常的有必要的&#xff0c;下面壹基比小鑫带大家一起去做个具体的探讨吧。 对于服务器不太了解的都认为&#xff0…...

vue(element ui安装)

目录 一&#xff0c;element ui安装二&#xff0c;main.js三&#xff0c;使用element ui最后 一&#xff0c;element ui安装 先在盘服中找到你创建的node的位置 如有不懂根据可以看看上一章安装node 然后在终端找到 进入这个位置之后就可以安装了 输入npm i element-ui -S这个…...

基于FPGA视频接口之HDMI2.0编/解码

简介 为什么要特别说明HDMI的版本,是因为HDMI的版本众多,代表的HDMI速度同样不同,当前版本在HDMI2.1速度达到48Gbps,可以传输4K及以上图像,但我们当前还停留在1080P@60部分,且使用的芯片和硬件结构有很大差别,故将HDMI分为两个部分说明1080@60以下分辨率和4K以上分辨率(…...

Codeforces Round #894 (Div.3)

文章目录 前言A. Gift Carpet题目&#xff1a;输入&#xff1a;输出&#xff1a;思路&#xff1a;代码&#xff1a; B. Sequence Game题目&#xff1a;输入&#xff1a;输出&#xff1a;思路&#xff1a;代码&#xff1a; C. Flower City Fence题目&#xff1a;输入&#xff1a…...

MyBatid动态语句且模糊查询

目录 什么是MyBtais动态语句&#xff1f;&#xff1f;&#xff1f; MyBatis常用的动态标签和表达式 if标签 Choose标签 where标签 MyBatis模糊查询 #与$的区别 ​编辑 MyBatis映射 resultType resultMap 什么是MyBtais动态语句&#xff1f;&#xff1f;&#xff1f;…...

JVM——垃圾回收器G1+垃圾回收调优

4.4 G1&#xff08;一个垃圾回收器&#xff09; 定义: 取代了CMS垃圾回收器。和CMS一样时并发的。 适用场景: 物理上分区&#xff0c;逻辑上分代。 相关JVM参数: -XX:UseG1GC-XX:G1HeapRegionSizesize-XX:MaxGCPauseMillistime 1) G1 垃圾回收阶段 三个回收阶段&#xff0…...

【SA8295P 源码分析】23 - QNX Ethernet MAC 驱动 之 emac1_config.conf 配置文件解析

【SA8295P 源码分析】23 - QNX Ethernet MAC 驱动 之 emac1_config.conf 配置文件解析 系列文章汇总见:《【SA8295P 源码分析】00 - 系列文章链接汇总》 本文链接:《【SA8295P 源码分析】23 - QNX Ethernet MAC 驱动 之 emac1_config.conf 配置文件解析》 主要参数如下: hw_…...

iptables的使用规则

环境中为了安全要限制swagger的访问&#xff0c;最简单的方式是通过iptables防火墙设置规则限制。 在测试服务器中设置访问swagger-ui.html显示如下&#xff0c;区分大小写&#xff1a; iptables设置限制访问9783端口的swagger字段的请求&#xff1a; iptables -A INPUT -p t…...

JS 动画 vs CSS 动画:究竟有何不同?

在 Web 前端开发中&#xff0c;动画是提高用户体验的关键因素之一。我们通常可以使用 JavaScript&#xff08;JS&#xff09;和 CSS 来创建动画效果。但是&#xff0c;这两者之间有哪些区别呢&#xff1f;在本文中&#xff0c;我们将深入研究 JS 动画和 CSS 动画&#xff0c;探…...

供应链 | 大数据报童模型:基于机器学习的实践见解

论文解读&#xff1a;李欣 马玺渊 作者&#xff1a;Gah-Yi Ban, Cynthia Rudin 引用&#xff1a;Ban, Gah-Yi and Cynthia Rudin. The big data newsvendor: Practical insights from machine learning. Operations Research 67.1 (2019): 90-108. 文章链接&#xff1a;https…...

docker详细操作--未完待续

docker介绍 docker官网: Docker&#xff1a;加速容器应用程序开发 harbor官网&#xff1a;Harbor - Harbor 中文 使用docker加速器: Docker镜像极速下载服务 - 毫秒镜像 是什么 Docker 是一种开源的容器化平台&#xff0c;用于将应用程序及其依赖项&#xff08;如库、运行时环…...

2021-03-15 iview一些问题

1.iview 在使用tree组件时&#xff0c;发现没有set类的方法&#xff0c;只有get&#xff0c;那么要改变tree值&#xff0c;只能遍历treeData&#xff0c;递归修改treeData的checked&#xff0c;发现无法更改&#xff0c;原因在于check模式下&#xff0c;子元素的勾选状态跟父节…...

IoT/HCIP实验-3/LiteOS操作系统内核实验(任务、内存、信号量、CMSIS..)

文章目录 概述HelloWorld 工程C/C配置编译器主配置Makefile脚本烧录器主配置运行结果程序调用栈 任务管理实验实验结果osal 系统适配层osal_task_create 其他实验实验源码内存管理实验互斥锁实验信号量实验 CMISIS接口实验还是得JlINKCMSIS 简介LiteOS->CMSIS任务间消息交互…...

多模态大语言模型arxiv论文略读(108)

CROME: Cross-Modal Adapters for Efficient Multimodal LLM ➡️ 论文标题&#xff1a;CROME: Cross-Modal Adapters for Efficient Multimodal LLM ➡️ 论文作者&#xff1a;Sayna Ebrahimi, Sercan O. Arik, Tejas Nama, Tomas Pfister ➡️ 研究机构: Google Cloud AI Re…...

select、poll、epoll 与 Reactor 模式

在高并发网络编程领域&#xff0c;高效处理大量连接和 I/O 事件是系统性能的关键。select、poll、epoll 作为 I/O 多路复用技术的代表&#xff0c;以及基于它们实现的 Reactor 模式&#xff0c;为开发者提供了强大的工具。本文将深入探讨这些技术的底层原理、优缺点。​ 一、I…...

聊一聊接口测试的意义有哪些?

目录 一、隔离性 & 早期测试 二、保障系统集成质量 三、验证业务逻辑的核心层 四、提升测试效率与覆盖度 五、系统稳定性的守护者 六、驱动团队协作与契约管理 七、性能与扩展性的前置评估 八、持续交付的核心支撑 接口测试的意义可以从四个维度展开&#xff0c;首…...

【碎碎念】宝可梦 Mesh GO : 基于MESH网络的口袋妖怪 宝可梦GO游戏自组网系统

目录 游戏说明《宝可梦 Mesh GO》 —— 局域宝可梦探索Pokmon GO 类游戏核心理念应用场景Mesh 特性 宝可梦玩法融合设计游戏构想要素1. 地图探索&#xff08;基于物理空间 广播范围&#xff09;2. 野生宝可梦生成与广播3. 对战系统4. 道具与通信5. 延伸玩法 安全性设计 技术选…...

CSS设置元素的宽度根据其内容自动调整

width: fit-content 是 CSS 中的一个属性值&#xff0c;用于设置元素的宽度根据其内容自动调整&#xff0c;确保宽度刚好容纳内容而不会超出。 效果对比 默认情况&#xff08;width: auto&#xff09;&#xff1a; 块级元素&#xff08;如 <div>&#xff09;会占满父容器…...

C++:多态机制详解

目录 一. 多态的概念 1.静态多态&#xff08;编译时多态&#xff09; 二.动态多态的定义及实现 1.多态的构成条件 2.虚函数 3.虚函数的重写/覆盖 4.虚函数重写的一些其他问题 1&#xff09;.协变 2&#xff09;.析构函数的重写 5.override 和 final关键字 1&#…...

第7篇:中间件全链路监控与 SQL 性能分析实践

7.1 章节导读 在构建数据库中间件的过程中&#xff0c;可观测性 和 性能分析 是保障系统稳定性与可维护性的核心能力。 特别是在复杂分布式场景中&#xff0c;必须做到&#xff1a; &#x1f50d; 追踪每一条 SQL 的生命周期&#xff08;从入口到数据库执行&#xff09;&#…...