README

workbunny/webman-rqueue

🐇 基于 Redis Stream 的轻量级队列，适用于 Webman 插件。 🐇

基于 Redis Stream 的轻量级队列，适用于 Webman 插件

常见问题

什么时候使用消息队列？

当你需要对系统进行解耦、削峰、异步处理时；如发送短信验证码、秒杀活动、资产的异步分账清算等。
RabbitMQ 和 Redis 的区别？

Redis 中的 Stream 的特性同样适用于消息队列，并且也包含了比较完善的 ACK 机制，但在一些点上与 RabbitMQ 存在不同：
- Redis Stream 没有完善的后台管理；RabbitMQ 拥有较为完善的后台管理及 Api；
- Redis 的持久化策略取舍：默认的 RDB 策略极端情况下存在丢失数据，AOF 策略则需要牺牲一些性能；RabbitMQ 持久化方案更多，可对消息持久化也可对队列持久化；
- RabbitMQ 拥有更多的插件可以提供更完善的协议支持及功能支持；
什么时候使用 Redis？什么时候使用 RabbitMQ？

当你的队列使用比较单一或者比较轻量的时候，请选用 Redis Stream；当你需要一个比较完整的消息队列体系，包括需要利用交换机来绑定不同队列做一些比较复杂的消息任务的时候，请选择 RabbitMQ；

当然，如果你的队列使用也比较单一，但你需要用到一些管理后台相关系统化的功能的时候，又不想花费太多时间去开发的时候，也可以使用 RabbitMQ；因为 RabbitMQ 提供了一整套后台管理的体系及 HTTP API 供开发者兼容到自己的管理后台中，不需要再消耗多余的时间去开发功能；

注：这里的轻量指的是 无需将应用中的队列服务独立化，该队列服务是该应用独享的

简介

基于 Redis Stream 的轻量级队列；
Queue 模式：多个消费者竞争消费
- 支持普通消费
- 支持延迟消费
Group 模式：多个消费组订阅消费
- 支持普通消费
- 支持延迟消费
Adaptive 模式：Queue 的自适应版消费者
- 支持普通消费
- 支持延迟消费
可靠的重载机制，防止消息意外丢失/重复消费
- 使用本地 sqlite 库储存因意外中断的消息
- 自动加载本地消息至队列
简单易用容易理解的使用方式
- 丰富的命令行助手，助手函数
- BuilderClass 继承模式（类似 ORM 的 ModelClass）
- 支持自定义 BuilderClass，实现自定义的消费逻辑

安装

环境依赖

php >=8.0
webman >= 1.0
redis >= 6.2

composer require workbunny/webman-rqueue

使用

QueueBuilder

一个 QueueBuilder 类对应一个消费 Group 和一个消费逻辑 QueueBuilder::handler()
一个 QueueBuilder 可对应一个/多个 Redis-Stream-Key，通过配置 QueueBuilder::$config['queues']
QueueBuilder 类使用定时器进行消费，每一次消费之后会根据消息的属性 _header['_delete'] 来进行消息释放

命令行

--mode 默认为 queue

创建

# 创建一个拥有单进程消费者的QueueBuilder
./webman workbunny:rqueue-builder testQueue --mode=queue
./webman workbunny:rqueue-builder testQueue -m queue
# 创建一个拥有4进程消费者的QueueBuilder
./webman workbunny:rqueue-builder testQueue 4 --mode=queue
./webman workbunny:rqueue-builder testQueue 4 -m queue
# 创建一个拥有单进程消费者的延迟QueueBuilder
./webman workbunny:rqueue-builder testQueue --delayed --mode=queue
./webman workbunny:rqueue-builder testQueue -dm queue
# 创建一个拥有4进程消费者的延迟QueueBuilder
./webman workbunny:rqueue-builder testQueue 4 --delayed --mode=queue
./webman workbunny:rqueue-builder testQueue 4 -dm queue

# 在 process/workbunny/rqueue 目录下创建 TestQueueBuilder.php
./webman workbunny:rqueue-builder testQueue
# 在 process/workbunny/rqueue/project 目录下创建 TestQueueBuilder.php
./webman workbunny:rqueue-builder project/testQueue
# 在 process/workbunny/rqueue/project 目录下创建 TestAllQueueBuilder.php
./webman workbunny:rqueue-builder project/testAllQueue
# 延迟同理

移除

移除包含了类文件的移除和配置的移除

# 移除Builder
./webman workbunny:rqueue-remove testQueue
# 移除延迟Builder
./webman workbunny:rqueue-remove testQueue --delayed
./webman workbunny:rqueue-remove testQueue -d
# 二级菜单同理

开启

开启仅对配置进行移除

# 开启Builder
./webman workbunny:rqueue-builder test --open --mode=queue
./webman workbunny:rqueue-builder test -om queue
# 开启延迟Builder
./webman workbunny:rqueue-builder test --open --delayed --mode=queue
./webman workbunny:rqueue-builder test -odm queue

# 二级菜单同理

关闭

关闭仅对配置进行移除

# 关闭Builder
./webman workbunny:rqueue-remove test --close --mode=queue
./webman workbunny:rqueue-remove test -cm queue
# 关闭延迟Builder
./webman workbunny:rqueue-remove test --close --delayed --mode=queue
./webman workbunny:rqueue-remove test -cdm queue

# 二级菜单同理

GroupBuilder

一个 GroupBuilder 类对应一个消费 Group 和一个消费逻辑 QueueBuilder::handler()
一个 GroupBuilder 可对应一个/多个 Redis-Stream-Key，通过配置 QueueBuilder::$config['queues']
GroupBuilder 类使用定时器进行消费，使用定时器释放当前 Stream-Key 上所有 Group 收取过的闲置消息
可以使用多个 GroupBuilder 类配置相同的 QueueBuilder::$config['queues']，从而达到一条/多条队列由不同的消费逻辑进行处理；
- 基于此特性，可以实现消息持久化的发布订阅
- 基于此特性，可以实现 RabbitMQ 的 exchange 模式

命令行

创建

# 创建一个拥有单进程消费者的GroupBuilder
./webman workbunny:rqueue-builder testGroup --mode=group
./webman workbunny:rqueue-builder testGroup -m group
# 创建一个拥有4进程消费者的GroupBuilder
./webman workbunny:rqueue-builder testGroup 4 --mode=group
./webman workbunny:rqueue-builder testGroup 4 -m group
# 创建一个拥有单进程消费者的延迟GroupBuilder
./webman workbunny:rqueue-builder testGroup --delayed--mode=group
./webman workbunny:rqueue-builder testGroup -dm group
# 创建一个拥有4进程消费者的延迟GroupBuilder
./webman workbunny:rqueue-builder testGroup 4 --delayed--mode=group
./webman workbunny:rqueue-builder testGroup 4 -dm group

# 二级菜单

# 在 process/workbunny/rqueue 目录下创建 TestGroupBuilder.php
./webman workbunny:rqueue-builder testGroup --mode=group
./webman workbunny:rqueue-builder testGroup -m group
# 在 process/workbunny/rqueue/project 目录下创建 TestGroupBuilder.php
./webman workbunny:rqueue-builder project/testGroup --mode=group
./webman workbunny:rqueue-builder project/testGroup -m group
# 在 process/workbunny/rqueue/project 目录下创建 TestAllGroupBuilder.php
./webman workbunny:rqueue-builder project/testAllGroup --mode=group
./webman workbunny:rqueue-builder project/testAllGroup -m group

移除

移除包含了类文件的移除和配置的移除

# 移除Builder
./webman workbunny:rqueue-remove testGroup --mode=group
./webman workbunny:rqueue-remove testGroup -m group
# 移除延迟Builder
./webman workbunny:rqueue-remove testGroup --delayed --mode=group
./webman workbunny:rqueue-remove testGroup -dm group
# 二级菜单同理

开启

开启仅对配置进行移除

# 开启Builder
./webman workbunny:rqueue-builder testGroup --open --mode=group
./webman workbunny:rqueue-builder testGroup -om group
# 开启延迟Builder
./webman workbunny:rqueue-builder testGroup --open --delayed --mode=group
./webman workbunny:rqueue-builder testGroup -odm group
# 二级菜单同理

关闭

关闭仅对配置进行移除

# 关闭Builder
./webman workbunny:rqueue-remove testGroup --close --mode=group
./webman workbunny:rqueue-remove testGroup -cm group
# 关闭延迟Builder
./webman workbunny:rqueue-remove testGroup --close --delayed --mode=group
./webman workbunny:rqueue-remove testGroup -cdm group
# 二级菜单同理

AdaptiveBuilder

说明

AdaptiveBuilder 与 QueueBuilder 的消费方式一致
AdaptiveBuilder 底层定时器会根据 闲置阈值 进行判断，消费定时器根据 退避指数 x 当前消费间隔 进行重置消费间隔

相较于其他 Builder 的优势

在消息负载较高的情况下，AdaptiveBuilder 是普通的 QueueBuilder
在消息负载较低的情况下，AdaptiveBuilder 根据闲置阈值对消费者的消费查询速率进行自适应退避调整，有效减少 redis 的查询压力
在延时消费场景下，AdaptiveBuilder 能有效减少因频繁查询 redis 而造成的 redis-server CPU 占用率较高的问题
在普通消费模式下，AdaptiveBuilder 相比于其他 Builder 能更快启动下一个消费周期，无需等待 Timer 的下一个 loop，消费更及时

命令行

创建

# 创建一个拥有单进程消费者的AdaptiveBuilder
./webman workbunny:rqueue-builder testAdaptive --mode=adaptive
./webman workbunny:rqueue-builder testAdaptive -m adaptive
# 创建一个拥有4进程消费者的AdaptiveBuilder
./webman workbunny:rqueue-builder testAdaptive 4 --mode=adaptive
./webman workbunny:rqueue-builder testAdaptive 4 -m adaptive
# 创建一个拥有单进程消费者的延迟AdaptiveBuilder
./webman workbunny:rqueue-builder testAdaptive --delayed--mode=adaptive
./webman workbunny:rqueue-builder testAdaptive -dm adaptive
# 创建一个拥有4进程消费者的延迟AdaptiveBuilder
./webman workbunny:rqueue-builder testAdaptive 4 --delayed--mode=adaptive
./webman workbunny:rqueue-builder testAdaptive 4 -dm adaptive

# 二级菜单

# 在 process/workbunny/rqueue 目录下创建
./webman workbunny:rqueue-builder testAdaptive --mode=adaptive
./webman workbunny:rqueue-builder testAdaptive -m adaptive
# 在 process/workbunny/rqueue/project 目录下创建 
./webman workbunny:rqueue-builder project/testAdaptive --mode=adaptive
./webman workbunny:rqueue-builder project/testAdaptive -m adaptive
# 在 process/workbunny/rqueue/project 目录下创建 
./webman workbunny:rqueue-builder project/testAllAdaptive --mode=adaptive
./webman workbunny:rqueue-builder project/testAllAdaptive -m adaptive

移除

移除包含了类文件的移除和配置的移除

# 移除Builder
./webman workbunny:rqueue-remove testAdaptive --mode=adaptive
./webman workbunny:rqueue-remove testAdaptive -m adaptive
# 移除延迟Builder
./webman workbunny:rqueue-remove testAdaptive --delayed --mode=adaptive
./webman workbunny:rqueue-remove testAdaptive -dm adaptive
# 二级菜单同理

开启

开启仅对配置进行移除

# 开启Builder
./webman workbunny:rqueue-builder testAdaptive --open --mode=adaptive
./webman workbunny:rqueue-builder testAdaptive -om adaptive
# 开启延迟Builder
./webman workbunny:rqueue-builder testAdaptive --open --delayed --mode=adaptive
./webman workbunny:rqueue-builder testAdaptive -odm adaptive
# 二级菜单同理

关闭

关闭仅对配置进行移除

# 关闭Builder
./webman workbunny:rqueue-remove testAdaptive --close --mode=adaptive
./webman workbunny:rqueue-remove testAdaptive -cm adaptive
# 关闭延迟Builder
./webman workbunny:rqueue-remove testAdaptive --close --delayed --mode=adaptive
./webman workbunny:rqueue-remove testAdaptive -cdm adaptive
# 二级菜单同理

自定义 Builder

如 queue/group Builder 都不满足需求，您可继承 AbstractBuilder 自行实现您所需要的 Builder

您的 Builder 基类需要继承 AbstractBuilder 实现，可参考 QueueBuilder/GroupBuilder
- onWorkerStart 用于进程启动时的触发逻辑，这里一般使用 Timer 结合读取队列触发 callback 来实现消费队列
- onWorkerStop 用于进程停止时的回收动作
- onWorkerReload 用于进程重载时的触发动作，除非有特殊处理，通常置空

    /*
     * Builder 启动时
     *
     * @param Worker $worker
     * @return void
     */
    abstract public function onWorkerStart(Worker $worker): void;

    /**
     * Builder 停止时
     *
     * @param Worker $worker
     * @return void
     */
    abstract public function onWorkerStop(Worker $worker): void;

    /**
     * Builder 重加载时
     *
     * @param Worker $worker
     * @return void
     */
    abstract public function onWorkerReload(Worker $worker): void;

classContent 方法是配合命令行，用于自动生成队列文件，如不使用，可置空

    /**
     * Command 获取需要创建的类文件内容
     *
     * @param string $namespace
     * @param string $className
     * @param bool $isDelay
     * @return string
     */
    abstract public static function classContent(string $namespace, string $className, bool $isDelay): string;

Traits 类提供所需的基础方法，按需在您的 Builder 基类中引用
- MessageQueueMethod 提供针对 redis-stream 队列的基础操作，如 ack、publish、consume 等、
- MessageTempMethod 提供本地数据缓存，用于对异常数据的收集和 requeue
- 如上述 traits 无法满足需求，可自定义 Traits

注意

QueueBuilder 与 GroupBuilder 在命令行自动生成时没有做类似 Delayed 的区分，用户可自行进行命名区分，如：

# 创建一个GroupBuilder
./webman workbunny:rqueue-builder testGroup --mode=group
# 创建一个QueueBuilder
./webman workbunny:rqueue-builder testQueue --mode=queue

创建的 Builder 类可以手动修改调整
为 Builder 添加进 process.php 的配置可以手动修改

查看 Builder

./webman workbunny:rqueue-list

注：当 Builder 未启动时，handler 与 count 显示为 --

+----------+-----------------------------------------------------------------------+-------------------------------------------------+-------+-------+
| name     | file                                                                  | handler                                         | count | mode  |
+----------+-----------------------------------------------------------------------+-------------------------------------------------+-------+-------+
| test     | /var/www/your-project/process/workbunny/rqueue/TestBuilder.php        | process\workbunny\rqueue\TestBuilder            | 1     | queue |
| test -d  | /var/www/your-project/process/workbunny/rqueue/TestBuilderDelayed.php | process\workbunny\rqueue\TestBuilderDelayed     | 1     | group |
+----------+-----------------------------------------------------------------------+-------------------------------------------------+-------+-------+

生产

发布普通消息

注：向普通队列发布延迟消息会抛出一个 WebmanRqueueException 异常

use function Workbunny\WebmanRqueue\sync_publish;
use function Workbunny\WebmanRqueue\sync_publish_get_ids;
use process\workbunny\rqueue\TestBuilder;

# 使用函数发布，返回受影响条数，多队列不具备事务一致
/** headers参数详见 @link Header */
sync_publish(TestBuilder::instance(), 'abc', [
	'_delete' => false
]);

# 使用对象发布，返回受影响条数，多队列不具备事务一致
/** headers参数详见 @link Header */
TestBuilder::instance()->publish('abc', [
	'_delete' => false
]);

# 返回消息ID组(数组)，多队列不具备事务一致
sync_publish_get_ids(TestBuilder::instance(), 'abc', [
	'_delete' => false
]);

# 返回消息ID组(数组)，多队列不具备事务一致
TestBuilder::instance()->publishGetIds('abc', [
	'_delete' => false
]);

发布延迟消息

注：向延迟队列发布普通消息会抛出一个 WebmanRqueueException 异常 注：延迟队列发布消息不支持指定消息 id

use function Workbunny\WebmanRqueue\sync_publish;
use function Workbunny\WebmanRqueue\sync_publish_get_ids;
use process\workbunny\rqueue\TestBuilder;

# 延迟10ms，返回受影响条数，多队列不具备事务一致
sync_publish(TestBuilder::instance(), 'abc', [
	'_delay' => 10
]);

# 延迟10ms，返回受影响条数，多队列不具备事务一致
TestBuilder::instance()->publish('abc', [
	'_delay' => 10
]);

# 延迟10ms，返回消息ID组(数组)，多队列不具备事务一致
sync_publish_get_ids(TestBuilder::instance(), 'abc', [
	'_delay' => 10
]);

# 延迟10ms，返回消息ID组(数组)，多队列不具备事务一致
TestBuilder::instance()->publishGetIds('abc', [
	'_delay' => 10
]);

说明

生产可用，欢迎 issue 和 PR；
Redis Stream 本身没有 delayed 或 non-delayed 之分，组件代码将它们区分的原因是不希望 delayed 被滥用；开发者应该明确哪些消息是延迟的、哪些是立即的，并且明确体现，也方便维护，因为延迟消息过多会导致消息堆积，从而占用 Redis 过多的资源；
延迟队列是通过对消息的不断读取放回并且加以判断是否达到延迟触发时间来进行模拟延迟触发的效果，消息不支持指定 id，请在逻辑内自行实现；
Redis Stream 的持久化依赖 Redis 本身的持久化策略，在一定情况下 Redis Stream 也并非是可靠型的消息队列;关于持久化相关内容，请仔细阅读 Redis中文文档；
本地重载机制使用了 SQLite3，详见 src/Builders/Traits/MessageTempMethod

workbunny /webman-rqueue

维护者

详细信息

README

基于 Redis Stream 的轻量级队列，适用于 Webman 插件

常见问题

简介

安装

环境依赖

使用

QueueBuilder

命令行

GroupBuilder

命令行

AdaptiveBuilder

说明

相较于其他 Builder 的优势

命令行

自定义 Builder

注意

查看 Builder

生产

发布普通消息

发布延迟消息

说明