AMQP 组件

hyperf/amqp 是实现 AMQP 标准的组件,主要适用于对 RabbitMQ 的使用。

安装

  1. composer require hyperf/amqp

默认配置

配置 类型 默认值 备注
host string localhost Host
port int 5672 端口号
user string guest 用户名
password string guest 密码
vhost string / vhost
concurrent.limit int 0 同时消费的数量
pool object 连接池配置
params object 基本配置
  1. <?php
  2. return [
  3. 'default' => [
  4. 'host' => 'localhost',
  5. 'port' => 5672,
  6. 'user' => 'guest',
  7. 'password' => 'guest',
  8. 'vhost' => '/',
  9. 'concurrent' => [
  10. 'limit' => 1,
  11. ],
  12. 'pool' => [
  13. 'min_connections' => 1,
  14. 'max_connections' => 10,
  15. 'connect_timeout' => 10.0,
  16. 'wait_timeout' => 3.0,
  17. 'heartbeat' => -1,
  18. ],
  19. 'params' => [
  20. 'insist' => false,
  21. 'login_method' => 'AMQPLAIN',
  22. 'login_response' => null,
  23. 'locale' => 'en_US',
  24. 'connection_timeout' => 3.0,
  25. 'read_write_timeout' => 3.0,
  26. 'context' => null,
  27. 'keepalive' => false,
  28. 'heartbeat' => 3,
  29. ],
  30. ],
  31. 'pool2' => [
  32. ...
  33. ]
  34. ];

可在 producer 或者 consumer__construct 函数中, 设置不同 pool.

投递消息

使用 gen:producer 命令创建一个 producer

  1. php bin/hyperf.php gen:amqp-producer DemoProducer

在 DemoProducer 文件中,我们可以修改 @Producer 注解对应的字段来替换对应的 exchangeroutingKey。其中 payload 就是最终投递到消息队列中的数据,所以我们可以随意改写 __construct 方法,只要最后赋值 payload 即可。示例如下。

使用 @Producer 注解时需 use Hyperf\Amqp\Annotation\Producer; 命名空间;

  1. <?php
  2. declare(strict_types=1);
  3. namespace App\Amqp\Producers;
  4. use Hyperf\Amqp\Annotation\Producer;
  5. use Hyperf\Amqp\Message\ProducerMessage;
  6. use App\Models\User;
  7. /**
  8. * DemoProducer
  9. * @Producer(exchange="hyperf", routingKey="hyperf")
  10. */
  11. class DemoProducer extends ProducerMessage
  12. {
  13. public function __construct($id)
  14. {
  15. // 设置不同 pool
  16. $this->poolName = 'pool2';
  17. $user = User::where('id', $id)->first();
  18. $this->payload = [
  19. 'id' => $id,
  20. 'data' => $user->toArray()
  21. ];
  22. }
  23. }

通过 DI Container 获取 Hyperf\Amqp\Producer 实例,即可投递消息。以下实例直接使用 ApplicationContext 获取 Hyperf\Amqp\Producer 其实并不合理,DI Container 具体使用请到 依赖注入 章节中查看。

<?php
use Hyperf\Amqp\Producer;
use App\Amqp\Producers\DemoProducer;
use Hyperf\Utils\ApplicationContext;

$message = new DemoProducer(1);
$producer = ApplicationContext::getContainer()->get(Producer::class);
$result = $producer->produce($message);

消费消息

使用 gen:amqp-consumer 命令创建一个 consumer

php bin/hyperf.php gen:amqp-consumer DemoConsumer

在 DemoConsumer 文件中,我们可以修改 @Consumer 注解对应的字段来替换对应的 exchangeroutingKeyqueue。其中 $data 就是解析后的消息数据。示例如下。

使用 @Consumer 注解时需 use Hyperf\Amqp\Annotation\Consumer; 命名空间;

<?php

declare(strict_types=1);

namespace App\Amqp\Consumers;

use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;

/**
 * @Consumer(exchange="hyperf", routingKey="hyperf", queue="hyperf", nums=1)
 */
class DemoConsumer extends ConsumerMessage
{
    public function consume($data): string
    {
        print_r($data);
        return Result::ACK;
    }
}

禁止消费进程自启

默认情况下,使用了 @Consumer 注解后,框架会自动创建子进程启动消费者,并且会在子进程异常退出后,重新拉起。如果出于开发阶段,进行消费者调试时,可能会因为消费其他消息而导致调试不便。

这种情况,只需要在 @Consumer 注解中配置 enable=false (默认为 true 跟随服务启动)或者在对应的消费者中重写类方法 isEnable() 返回 false 即可

<?php

declare(strict_types=1);

namespace App\Amqp\Consumers;

use Hyperf\Amqp\Annotation\Consumer;
use Hyperf\Amqp\Message\ConsumerMessage;
use Hyperf\Amqp\Result;

/**
 * @Consumer(exchange="hyperf", routingKey="hyperf", queue="hyperf", nums=1, enable=false)
 */
class DemoConsumer extends ConsumerMessage
{
    public function consume($data): string
    {
        print_r($data);
        return Result::ACK;
    }

    public function isEnable(): bool
    {
        return parent::isEnable();
    }
}

消费结果

框架会根据 Consumer 内的 consume 方法所返回的结果来决定该消息的响应行为,共有 4 中响应结果,分别为 \Hyperf\Amqp\Result::ACK\Hyperf\Amqp\Result::NACK\Hyperf\Amqp\Result::REQUEUE\Hyperf\Amqp\Result::DROP,每个返回值分别代表如下行为:

返回值 行为
\Hyperf\Amqp\Result::ACK 确认消息正确被消费掉了
\Hyperf\Amqp\Result::NACK 消息没有被正确消费掉,以 basic_nack 方法来响应
\Hyperf\Amqp\Result::REQUEUE 消息没有被正确消费掉,以 basic_reject 方法来响应,并使消息重新入列
\Hyperf\Amqp\Result::DROP 消息没有被正确消费掉,以 basic_reject 方法来响应