Yet-another swoole 异步队列，支持弹性扩容，工做进程协程支持

Aint Queue

基于 Swoole 的一个异步队列库，可弹性伸缩的工做进程池，工做进程协程支持。

Github 地址：https://github.com/Littlesqx/...

特性

• 默认 Redis 驱动
• 秒级延时任务
• 自定义重试次数和时间
• 自定义错误回调
• 支持任务执行中间件
• 自定义队列快照事件
• 弹性多进程消费
• 工做进程协程支持
• 内置漂亮的仪表盘

环境

• PHP 7.2+
• Swoole 4.4+
• Redis 3.2+ (redis 驱动)

安装

$ composer require littlesqx/aint-queue -vvv

使用

配置

默认读取配置路径： config/aint-queue.php， 不存在时读取 /vendor/littlesqx/aint-queue/src/Config/config.php 。

<?php

use Littlesqx\AintQueue\Driver\Redis\Queue as RedisQueue;
use Littlesqx\AintQueue\Logger\DefaultLogger;

return [
    // channel_name => [...config]
    'default' => [
        'driver' => [
            'class' => RedisQueue::class,
            'connection' => [
                'host' => '127.0.0.1',
                'port' => 6379,
                'database' => '0',
                // 'password' => 'password',
            ],
            'pool_size' => 8,
            'pool_wait_timeout' => 1,
            'handle_timeout' => 60 * 30,
        ],
        'logger' => [
            'class' => DefaultLogger::class,
                'options' => [
                    'level' => \Monolog\Logger::DEBUG,
                ],
            ],
        'pid_path' => '/var/run/aint-queue',
        'consumer' => [
            'sleep_seconds' => 1,
            'memory_limit' => 96,
            'dynamic_mode' => true,
            'capacity' => 6,
            'flex_interval' => 5 * 60,
            'min_worker_number' => 5,
            'max_worker_number' => 30,
            'max_handle_number' => 0,
        ],
        'job_snapshot' => [
            'interval' => 5 * 60,
            'handler' => [],
        ],
    ],
];

全部参数：

name	type	comment	default
channel	string	频道。队列的单位，每一个频道内的消息对应着各自的消费者和生产者。支持多频道。在命令行使用 --channel 参数。	default
driver.class	string	队列驱动类，须要实现 QueueInterface。	Redis
driver.connection	map	驱动配置。
pid_path	string	主进程的 PID 文件存储路径。注意运行用户须要可读写权限。	/var/run/aint-queue
consumer.sleep_seconds	int	当任务空闲时，每次 pop 操做后的睡眠秒数。	1
consumer.memory_limit	int	工做进程的最大使用内存，超出则重启。单位 MB。	96
consumer.dynamic_mode	bool	是否开启自动伸缩工做进程。	true
consumer.capacity	int	表明每一个工做进程在短期内而且健康状态下的最多处理消息数，它影响了工做进程的自动伸缩策略。	5
consumer.flex_interval	int	每 flex_interval 秒，监控进程尝试调整工做进程数（假设开启了自动伸缩工做进程）。	5
consumer.min_worker_number	int	工做进程最小数目。	5
consumer.max_worker_number	int	工做进程最大数目。	30
consumer.max_handle_number	int	当前工做进程最大处理消息数，超出后重启。0 表明无限制。	0
job_snapshot	map	每隔 job_snapshot.interval 秒，job_snapshot.handles 会被依次执行。job_snapshot.handles 须要实现 JobSnapshotterInterface。

消息推送

能够在 cli/fpm 运行模式下使用：

<?php

use Littlesqx\AintQueue\Driver\DriverFactory;

$queue = DriverFactory::make($channel, $options);

// push a job
$queue->push(function () {
    echo "Hello aint-queue\n";
});

// push a delay job
$closureJob = function () {
    echo "Hello aint-queue delayed\n";
};
$queue->push($closureJob, 5);

更建议使用类任务，这样功能上会更加完整，也能够得到更好的编码体验和性能。

• 建立的任务类须要实现 JobInterface，详细可参考 /example

  • handle(): void: 任务主体，你要执行的内容，在这里能够使用 swoole 的相关 api（好比建立协程等）；
  • canRetry(int $attempt, $error): bool: 决定是否要重试；
  • retryAfter(int $attempt): int: 决定重试延时时间；
  • failed(int $id, array $payload): void: 任务完全失败了（就是达到了重试次数还没成功）；
  • middleware(): array: 当前任务的中间件列表，原理参考 laravel pipeline 流水线；
• 注意任务必须能在生产者和消费者中（反）序列化，意味着须要在同一个项目
• 利用队列快照事件你能够实现队列实时监控，而利用任务中间件，你能够实现任务执行速率限制，任务执行日志等。

队列管理

推荐使用 Supervisor 等进程管理工具守护工做进程。

vendor/bin/aint-queue

AintQueue Console Tool

Usage:
  command [options] [arguments]

Options:
  -h, --help            Display this help message
  -q, --quiet           Do not output any message
  -V, --version         Display this application version
      --ansi            Force ANSI output
      --no-ansi         Disable ANSI output
  -n, --no-interaction  Do not ask any interactive question
  -v|vv|vvv, --verbose  Increase the verbosity of messages: 1 for normal output, 2 for more verbose output and 3 for debug

Available commands:
  help                 Displays help for a command
  list                 Lists commands
 queue
  queue:clear          Clear the queue.
  queue:dashboard      Start http server for dashboard.
  queue:reload-failed  Reload all the failed jobs onto the waiting queue.
  queue:status         Get the execute status of specific queue.
 worker
  worker:listen        Listen the queue.
  worker:reload        Reload worker for the queue.
  worker:run           Run the specific job.
  worker:stop          Stop listening the queue.

测试

composer test