Download the PHP package easyswoole/queue without Composer

Queue

EasySwoole Queue 提供普通任务、延迟任务和需要确认的可信任务。当前提供 Redis 和 RabbitMQ 驱动，并内置连接池；Redis 驱动同时支持单机和 Redis Cluster。

安装

运行环境要求：

• PHP 8.0+
• Swoole 5.1+
• phpredis 扩展
• 协程环境

phpredis 是阻塞客户端，在启动协程前需要启用 TCP Hook：

Redis 单机配置

Redis Cluster 配置

Cluster 模式下，同一个 topic 使用的所有 Redis key 都带有相同的 hash tag，因此 Lua 脚本涉及的 key 会落在同一个 slot。

Redis 驱动配置项

连接池还可以使用 minObjectNum、maxObjectNum、getObjectTimeout 等 EasySwoole\Pool\Config 配置项。

RabbitMQ 配置

RabbitMQ 驱动基于 php-amqplib/php-amqplib。该客户端内部使用 stream_select()， 因此必须在协程启动前同时开启 TCP Hook 和 Stream Hook：

在提供 SWOOLE_HOOK_STREAM_SELECT 常量的 Swoole 版本中，它与 SWOOLE_HOOK_STREAM_FUNCTION 表示相同的 Stream Hook。

RabbitMQ 驱动使用一个独立的后台消费长连接，通过 basic_consume 接收消息，并把任务分发到本地协程缓冲区。普通任务进入本地缓冲区后立即 ACK，可信任务则等待业务调用 confirm()。ACK、确认超时和重投都由拥有原 Channel 的后台协程统一执行，因此每条可信任务不会单独占用连接；同一进程内同时处理的可信任务数量主要由 prefetchCount 控制。

发布和消费分别使用连接池：发布池隔离多个业务协程的并发写入，消费池固定为 minObjectNum=1、maxObjectNum=1，其长连接不占用发布池配额。可信任务确认超时后，驱动会先用 Publisher Confirm 发布带有最新 runTimes 的任务，再 ACK 原消息。进程或连接在确认前异常中断时，RabbitMQ 会重新投递尚未 ACK 的原消息；业务仍需保持幂等。

可信任务场景建议从默认的 prefetchCount=100 开始调优。该值过大会让消费协程连续接收过多未确认消息，增加业务协程提交 confirm() 的等待时间；应根据单进程实际并发任务数和处理耗时逐步调整。

生产环境建议启用非零 heartbeat，让连接池健康检查能够发现静默断线和 TCP 半开连接。php-amqplib 要求 readWriteTimeout 至少为 heartbeat 的两倍，例如：

heartbeat=0 时仍会检查连接和 Channel 的本地状态，但无法通过心跳发现静默断线。

RabbitMQ 延迟任务使用带消息 TTL 的延迟队列和 Dead Letter Exchange 实现。RabbitMQ 只会检查队首消息是否过期，因此同一延迟队列中混用不同延迟时间时，短延迟消息可能被前面的长延迟消息阻塞；需要严格定时精度的场景应使用固定 TTL 队列或延迟消息插件。

普通任务

普通任务的 retryTimes 默认为 0。任务被成功取出后即从 Redis 删除，不需要调用 confirm()。

pop($timeout) 在队列为空时使用协程短轮询等待，每次轮询结束都会归还 Redis 连接，不会在整个等待期间独占连接池连接。

延迟任务

任务在延迟时间到达后才能被消费者取出。普通延迟任务和可信延迟任务分别存储，是否可信仍由 retryTimes 决定。

可信任务

当 retryTimes 大于 0 时，任务使用“至少投递一次”语义：

• pop() 会原子地将任务从待消费队列转移到处理中集合。
• 在 waitConfirmTime 时间内，任务不会被重复领取。
• 消费成功后必须确认 pop() 返回的 Job。
• 到期仍未确认时，任务会被重新放回待消费队列。
• 每次重新投递都会增加 runTimes，达到 retryTimes 后不再投递。

可信任务可能被重复投递，业务处理逻辑必须保证幂等。消费回执由驱动写入 Job，业务代码不需要设置 queueReceiptId。

持续消费

CLI 完整示例

测试

Redis 集成测试会为每个用例创建唯一 topic，并在结束后自动清理对应 key。详细变量说明见 tests/README.md。

单机 Redis 示例：

当前测试覆盖：

• 普通任务 push/pop。
• 可信任务领取与确认。
• waitConfirmTime 内不会重复领取，超时后重新投递。
• 延迟任务到期前不可见、到期后可消费。

性能与可靠性基准

以下结果于 2026-10-11 在一台阿里云 2 核 2GB 实例上测得。Redis、RabbitMQ、PHP 客户端和压测程序均运行在同一台实例，通过 127.0.0.1 通信，因此结果主要反映队列驱动及服务端处理能力，不包含公网延迟。

测试环境和口径：

• 2 vCPU，实例标称 2GB 内存，操作系统可见约 1.6GiB。
• PHP 8.4.26、Swoole 6.2.2。
• Redis 6.2.22，单机模式，RDB 开启、AOF 关闭。
• RabbitMQ 4.3.6，持久化队列和持久化消息，Publisher Confirm 开启。
• 发布连接池最大连接数为 20，同时使用 20 个发布协程和 20 个消费协程；RabbitMQ 另有一个固定 1/1 的消费连接池。
• RabbitMQ 使用 heartbeat=10、readWriteTimeout=25、prefetchCount=100。
• 每个 Job 的 jobData 为 256 字节；每种驱动和任务模式每轮处理 20,000 条任务，连续执行三轮，表中为三轮中位数。

可靠性检查结果：

• 共处理 240,000 条任务，其中可信任务 120,000 条；每轮均为 20,000 个唯一 Job、0 个重复、0 个错误，消费完成后队列数量为 0。
• 100 条 RabbitMQ 可信任务同时确认超时后全部重新投递，runTimes 均由 1 变为 2，重新确认后队列为空。
• 强制断开 RabbitMQ 消费连接后，旧回执全部确认失败；驱动自动重建消费连接，300 条任务全部恢复，测试中没有丢失或重复。
• 在 10,000 次并发发布期间由服务端关闭 21 个 RabbitMQ 连接，19 个在途 Publisher Confirm 明确抛出连接关闭异常，其余 9,981 条成功；连接池重建后追加 100 条全部成功，Broker 最终计数准确为 10,081，没有静默丢失。

这些数据用于展示上述固定环境下的参考性能，不代表不同硬件、持久化策略、消息大小和业务处理逻辑下的保证值。可信队列采用“至少投递一次”语义；网络在确认临界点中断时，调用方重试仍可能产生重复，因此业务处理必须保持幂等。

自定义驱动

自定义队列驱动需要实现 EasySwoole\Queue\AbstractInterface\QueueDriverInterface。驱动对象必须能够在初始化前被克隆；Queue 会为不同 topic 的 Producer 和 Consumer 分别克隆并调用一次 init()。