异步队列

异步队列区别于 RabbitMQ Kafka 等消息队列,它只提供一种 异步处理异步延时处理 的能力,并 不能 严格地保证消息的持久化和 不支持 ACK 应答机制。

安装

  1. composer require hyperf/async-queue

配置

配置文件位于 config/autoload/async_queue.php,如文件不存在可自行创建。

暂时只支持 Redis Driver 驱动。

配置 类型 默认值 备注
driver string Hyperf\AsyncQueue\Driver\RedisDriver::class
channel string queue 队列前缀
redis.pool string default redis 连接池
timeout int 2 pop 消息的超时时间
retry_seconds int,array 5 失败后重新尝试间隔
handle_timeout int 10 消息处理超时时间
processes int 1 消费进程数
concurrent.limit int 1 同时处理消息数
max_messages int 0 进程重启所需最大处理的消息数 默认不重启
  1. <?php
  2. return [
  3. 'default' => [
  4. 'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
  5. 'redis' => [
  6. 'pool' => 'default'
  7. ],
  8. 'channel' => 'queue',
  9. 'timeout' => 2,
  10. 'retry_seconds' => 5,
  11. 'handle_timeout' => 10,
  12. 'processes' => 1,
  13. 'concurrent' => [
  14. 'limit' => 5,
  15. ],
  16. ],
  17. ];

retry_seconds 也可以传入数组,根据重试次数相应修改重试时间,例如

  1. <?php
  2. return [
  3. 'default' => [
  4. 'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
  5. 'channel' => 'queue',
  6. 'retry_seconds' => [1, 5, 10, 20],
  7. 'processes' => 1,
  8. ],
  9. ];

使用

消费消息

组件已经提供了默认子进程,只需要将它配置到 config/autoload/processes.php 中即可。

  1. <?php
  2. return [
  3. Hyperf\AsyncQueue\Process\ConsumerProcess::class,
  4. ];

当然,您也可以将以下 Process 添加到自己的项目中。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Process;
  4. use Hyperf\AsyncQueue\Process\ConsumerProcess;
  5. use Hyperf\Process\Annotation\Process;
  6. /**
  7. * @Process(name="async-queue")
  8. */
  9. class AsyncQueueConsumer extends ConsumerProcess
  10. {
  11. }

生产消息

传统方式

这种模式会把对象直接序列化然后存到 Redis 等队列中,所以为了保证序列化后的体积,尽量不要将 ContainerConfig 等设置为成员变量。

比如以下 Job 的定义,是 不可取

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Job;
  4. use Hyperf\AsyncQueue\Job;
  5. use Psr\Container\ContainerInterface;
  6. class ExampleJob extends Job
  7. {
  8. public $container;
  9. public $params;
  10. public function __construct(ContainerInterface $container, $params)
  11. {
  12. $this->container = $container;
  13. $this->params = $params;
  14. }
  15. public function handle()
  16. {
  17. // 根据参数处理具体逻辑
  18. var_dump($this->params);
  19. }
  20. }
  21. $job = make(ExampleJob::class);

正确的 Job 应该是只有需要处理的数据,其他相关数据,可以在 handle 方法中重新获取,如下。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Job;
  4. use Hyperf\AsyncQueue\Job;
  5. class ExampleJob extends Job
  6. {
  7. public $params;
  8. public function __construct($params)
  9. {
  10. // 这里最好是普通数据,不要使用携带 IO 的对象,比如 PDO 对象
  11. $this->params = $params;
  12. }
  13. public function handle()
  14. {
  15. // 根据参数处理具体逻辑
  16. // 通过具体参数获取模型等
  17. var_dump($this->params);
  18. }
  19. }

正确定义完 Job 后,我们需要写一个专门投递消息的 Service,代码如下。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Service;
  4. use App\Job\ExampleJob;
  5. use Hyperf\AsyncQueue\Driver\DriverFactory;
  6. use Hyperf\AsyncQueue\Driver\DriverInterface;
  7. class QueueService
  8. {
  9. /**
  10. * @var DriverInterface
  11. */
  12. protected $driver;
  13. public function __construct(DriverFactory $driverFactory)
  14. {
  15. $this->driver = $driverFactory->get('default');
  16. }
  17. /**
  18. * 生产消息.
  19. * @param $params 数据
  20. * @param int $delay 延时时间 单位秒
  21. */
  22. public function push($params, int $delay = 0): bool
  23. {
  24. // 这里的 `ExampleJob` 会被序列化存到 Redis 中,所以内部变量最好只传入普通数据
  25. // 同理,如果内部使用了注解 @Value 会把对应对象一起序列化,导致消息体变大。
  26. // 所以这里也不推荐使用 `make` 方法来创建 `Job` 对象。
  27. return $this->driver->push(new ExampleJob($params), $delay);
  28. }
  29. }

投递消息

接下来,调用我们的 QueueService 投递消息即可。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Controller;
  4. use App\Service\QueueService;
  5. use Hyperf\Di\Annotation\Inject;
  6. use Hyperf\HttpServer\Annotation\AutoController;
  7. /**
  8. * @AutoController
  9. */
  10. class QueueController extends Controller
  11. {
  12. /**
  13. * @Inject
  14. * @var QueueService
  15. */
  16. protected $service;
  17. /**
  18. * 传统模式投递消息
  19. */
  20. public function index()
  21. {
  22. $this->service->push([
  23. 'group@hyperf.io',
  24. 'https://doc.hyperf.io',
  25. 'https://www.hyperf.io',
  26. ]);
  27. return 'success';
  28. }
  29. }

注解方式

框架除了传统方式投递消息,还提供了注解方式。

让我们重写上述 QueueService,直接将 ExampleJob 的逻辑搬到 example 方法中,并加上对应注解 AsyncQueueMessage,具体代码如下。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Service;
  4. use Hyperf\AsyncQueue\Annotation\AsyncQueueMessage;
  5. class QueueService
  6. {
  7. /**
  8. * @AsyncQueueMessage
  9. */
  10. public function example($params)
  11. {
  12. // 需要异步执行的代码逻辑
  13. var_dump($params);
  14. }
  15. }

投递消息

注解模式投递消息就跟平常调用方法一致,代码如下。

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Controller;
  4. use App\Service\QueueService;
  5. use Hyperf\Di\Annotation\Inject;
  6. use Hyperf\HttpServer\Annotation\AutoController;
  7. /**
  8. * @AutoController
  9. */
  10. class QueueController extends Controller
  11. {
  12. /**
  13. * @Inject
  14. * @var QueueService
  15. */
  16. protected $service;
  17. /**
  18. * 注解模式投递消息
  19. */
  20. public function example()
  21. {
  22. $this->service->example([
  23. 'group@hyperf.io',
  24. 'https://doc.hyperf.io',
  25. 'https://www.hyperf.io',
  26. ]);
  27. return 'success';
  28. }
  29. }

事件

事件名称 触发时机 备注
BeforeHandle 处理消息前触发
AfterHandle 处理消息后触发
FailedHandle 处理消息失败后触发
RetryHandle 重试处理消息前触发
QueueLength 每处理 500 个消息后触发 用户可以监听此事件,判断失败或超时队列是否有消息积压

QueueLengthListener

框架自带了一个记录队列长度的监听器,默认不开启,您如果需要,可以自行添加到 listeners 配置中。

  1. <?php
  2. declare(strict_types=1);
  3. return [
  4. Hyperf\AsyncQueue\Listener\QueueLengthListener::class
  5. ];

任务执行流转流程

任务执行流转流程主要包括以下几个队列:

队列名 备注
waiting 等待消费的队列
reserved 正在消费的队列
delayed 延迟消费的队列
failed 消费失败的队列
timeout 消费超时的队列 (虽然超时,但可能执行成功)

队列流转顺序如下:

  1. graph LR;
  2. A[投递延时消息]-->C[delayed队列];
  3. B[投递消息]-->D[waiting队列];
  4. C--到期-->D;
  5. D--消费-->E[reserved队列];
  6. E--成功-->F[删除消息];
  7. E--失败-->G[failed队列];
  8. E--超时-->H[timeout队列];

配置多个异步队列

当您需要使用多个队列来区分消费高频和低频或其他种类的消息时,可以配置多个队列。

  1. 添加配置
  1. <?php
  2. return [
  3. 'default' => [
  4. 'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
  5. 'channel' => '{queue}',
  6. 'timeout' => 2,
  7. 'retry_seconds' => 5,
  8. 'handle_timeout' => 10,
  9. 'processes' => 1,
  10. 'concurrent' => [
  11. 'limit' => 2,
  12. ],
  13. ],
  14. 'other' => [
  15. 'driver' => Hyperf\AsyncQueue\Driver\RedisDriver::class,
  16. 'channel' => '{other.queue}',
  17. 'timeout' => 2,
  18. 'retry_seconds' => 5,
  19. 'handle_timeout' => 10,
  20. 'processes' => 1,
  21. 'concurrent' => [
  22. 'limit' => 2,
  23. ],
  24. ],
  25. ];
  1. 添加消费进程
  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Process;
  4. use Hyperf\AsyncQueue\Process\ConsumerProcess;
  5. use Hyperf\Process\Annotation\Process;
  6. /**
  7. * @Process()
  8. */
  9. class ConsumerProcess extends ConsumerProcess
  10. {
  11. /**
  12. * @var string
  13. */
  14. protected $queue = 'other';
  15. }
  1. 调用
  1. use Hyperf\AsyncQueue\Driver\DriverFactory;
  2. use Hyperf\Utils\ApplicationContext;
  3. $driver = ApplicationContext::getContainer()->get(DriverFactory::class)->get('other');
  4. return $driver->push(new ExampleJob());

安全关闭

异步队列在终止时,如果正在进行消费逻辑,可能会导致出现错误。框架提供了 DriverStopHandler ,可以让异步队列进程安全关闭。

安装信号处理器

  1. composer require hyperf/signal

添加配置

  1. <?php
  2. declare(strict_types=1);
  3. return [
  4. 'handlers' => [
  5. Hyperf\AsyncQueue\Signal\DriverStopHandler::class,
  6. ],
  7. 'timeout' => 5.0,
  8. ];