README

Gaia框架基于redis的消息队列工具，基于workerman/redis-queue封装扩展。

功能特性

支持多Redis连接配置，多个队列连接相互独立
支持延迟消息、消费失败自动重试（线性退避）、超过重试次数进入失败队列
消费成功/失败回调，支持自定义结果驱动（内置文件日志、Mysql 落库两种）
消费统计基于Redis聚合，多进程部署下监控数据全局准确
内置失败队列治理指令：查看、统计、重试、清空
内置日志表清理指令，支持分批删除
脚手架指令，一键生成消费者模板

安装

composer require mongdch/gaia-queue

发布

php gaia vendor:publish gaia\queue

发布后框架目录结构：

bin/queue.php              # 队列应用启动入口
config/queue/              # 配置文件
support/queue/             # 进程、服务与指令
app/queue/                 # 消费者目录（示例消费者）

注意：重复执行发布会覆盖app/queue下的同名示例文件，自定义消费者请使用不同的文件名。

启动

php bin/queue.php start -d

配置说明

配置文件config/queue/app.php：

return [
    // 消费者进程存放目录路径
    'consumers_path'    => APP_PATH . DIRECTORY_SEPARATOR . 'queue' . DIRECTORY_SEPARATOR,
    // 消费者命名空间
    'namespace'         => '\\app\queue',
    // 回调处理驱动，空则不记录消费结果
    'handler_driver'    => gaia\queue\driver\Logger::class,
    // 日志表表名，驱动为mysql时有效
    'log_table'         => env('QUEUE_LOG_TABLE', 'queue_log'),
    // 日志时间格式，空则为时间戳，驱动为mysql时有效
    'log_time_format'   => env('QUEUE_LOG_TIME_FORMAT', 'Y-m-d H:i:s'),
    // 消息队列进程配置
    'process'           => [
        // 监听协议端口，采用text协议，用于监控通信
        'listen'        => 'text://127.0.0.1:7123',
        // 进程数，0则默认cpu数量
        'count'         => env('QUEUE_SERVER_COUNT', 0),
        // 是否允许进程重载
        'reloadable'    => true,
        // 其余参数见：user/group/reusePort/context/transport
    ],
    // 日志通道配置，见下文
    'log'               => [...],
];

配置文件config/queue/queue.php（Redis 连接配置）：

return [
    // 默认连接名
    'default'       => 'default',
    'connections'   => [
        'default'   => [
            'scheme'        => 'redis',
            'host'          => env('QUEUE_HOST', env('REDIS_HOST', '127.0.0.1')),
            'port'          => env('QUEUE_PORT', env('REDIS_PORT', 6379)),
            'auth'          => env('QUEUE_AUTH', env('REDIS_AUTH', '')),
            // 自定义键前缀，注意：生产者与消费者需使用相同前缀
            'prefix'        => env('QUEUE_PREFIX', env('REDIS_PREFIX', '')),
            'database'      => env('QUEUE_DB', 5),
            // 消费失败后重试次数
            'max_attempts'  => env('QUEUE_ATTEMPTS', 5),
            // 重试间隔基数，实际间隔为 retry_seconds * 已重试次数
            'retry_seconds' => env('QUEUE_RETRY', 5)
        ]
    ],
];

定义消费者

在app/queue目录下创建消费者类，实现gaia\queue\ConsumerInterface，或直接使用脚手架：

php gaia queue:make-queue demo
# 生成 app/queue/DemoQueue.php

class DemoQueue implements ConsumerInterface
{
    use Instance;

    // 链接的队列配置名，空则使用默认队列配置
    public function connection(): string
    {
        return 'default';
    }

    // 监听的队列名
    public function queue(): string
    {
        return 'demo-queue';
    }

    // 队列描述信息
    public function describe(): string
    {
        return '演示队列';
    }

    // 处理函数
    public function consume($data)
    {
        // 消费逻辑，正常返回即视为消费成功
        return '处理成功';
    }

    // （可选）消费失败回调，return 修改后的$package可调整重试数据
    public function onConsumeFailure(Throwable $e, array $package) {}

    // （可选）消费成功回调
    public function onConsumeSuccess($result, array $package) {}

    // （可选）封装发消息方法
    public function sendQuery(array $data = [], int $delay = 0): bool
    {
        return QueueService::syncSend($this->queue(), $data, $delay, $this->connection());
    }
}

失败与重试

consume() 抛出任意异常：按max_attempts自动重试，间隔为retry_seconds * 已重试次数，超过次数进入失败队列
抛出Workerman\RedisQueue\UnretryableException：不重试，直接进入失败队列（适用于参数错误等不可恢复场景）
consume() 正常返回（包括返回错误提示字符串）视为消费成功，不会重试，失败请抛出异常

投递消息

同步投递（推荐，任意环境可用）

基于RedisService直写 Redis，不依赖队列进程是否存活：

use support\queue\QueueService;

// 立即消费
QueueService::syncSend('demo-queue', ['id' => 1]);

// 延迟60秒消费
QueueService::syncSend('demo-queue', ['id' => 1], 60);

// 指定连接
QueueService::syncSend('demo-queue', ['id' => 1], 0, 'other');

异步投递（需要 workerman 环境）

在 workerman 运行环境中（如 gaia 的 http 进程内）使用异步 IO：

QueueService::asyncSend('demo-queue', ['id' => 1], 60);

syncSend与消费者通过 Redis key 交互，需保证两侧使用相同的连接配置与前缀。

内置消费者

发布后自带三个示例消费者：

类	队列名	说明
`DemoQueue`	`demo-queue`	演示用
`HttpQueue`	`send-http`	异步 HTTP 请求，请求失败默认抛异常触发重试（`$retryOnFailure`可关闭）
`EmailQueue`	`send-email`	异步发送邮件，依赖`app\service\MailerService`

消费结果驱动

通过配置handler_driver记录每次消费结果：

gaia\queue\driver\Logger::class：写入文件日志（通道与格式见log配置），成功 info / 失败 error
gaia\queue\driver\Mysql::class：写入 Mysql 日志表，需先初始化表结构
留空：不记录

自定义驱动实现gaia\queue\DriverInterface即可：

interface DriverInterface
{
    public function handeler(array $package, bool $status, $result = '');
}

指令列表

# 生成消费者模板
php gaia queue:make-queue <name>

# 初始化/升级日志表（幂等，可重复执行）
php gaia queue:dbinit

# 查看失败消息列表
php gaia queue:failed list [--connection=default] [--limit=20]

# 查看失败消息数量（可用于告警脚本）
php gaia queue:failed count [--connection=xxx]

# 重试失败消息（重置重试次数后重新投递）
php gaia queue:failed retry --index=3
php gaia queue:failed retry --all

# 清空失败队列
php gaia queue:failed flush [--force]

# 清理日志表，默认清理30天前的记录
php gaia queue:log-clear --days=30

监控

队列进程监听text://127.0.0.1:7123（见process.listen配置），可获取全局聚合的消费统计：

use support\queue\QueueService;

$pool = QueueService::getPool();
// [
//   [
//     'id' => 'default_demo-queue',
//     'connection' => 'default',
//     'queue' => 'demo-queue',
//     'describe' => '演示队列',
//     'success' => 100,            // 累计消费成功数（Redis聚合）
//     'failure' => 2,              // 累计消费失败数（含重试）
//     'last_running_time' => '...', // 最近消费时间
//     'create_time' => '...',       // 进程启动时间
//   ], ...
// ]

注意事项

消息语义为 at-most-once：消息被取出后若进程崩溃，该条消息会丢失，重要业务请在消费逻辑中自行保证幂等与补偿
消费回调在事件循环内同步执行，单个慢消费会阻塞同进程内的其他队列，建议将重 IO 任务放到独立连接/队列
Mysql 驱动逐条同步落库，高吞吐场景建议改用 Logger 驱动或自行扩展批量写入
失败队列与消费统计依赖 Redis 持久化，请勿随意 flush 对应 db

License

MIT

mongdch / gaia-queue

Package info

Security

Statistics