AMQP组件

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

安装

  1. composer require hyperf/amqp

默认配置

  1. <?php
  3. return [
  4.     'default' => [
  5.         'host' => 'localhost',
  6.         'port' => 5672,
  7.         'user' => 'guest',
  8.         'password' => 'guest',
  9.         'vhost' => '/',
  10.         'pool' => [
  11.             'min_connections' => 1,
  12.             'max_connections' => 10,
  13.             'connect_timeout' => 10.0,
  14.             'wait_timeout' => 3.0,
  15.             'heartbeat' => -1,
  16.         ],
  17.         'params' => [
  18.             'insist' => false,
  19.             'login_method' => 'AMQPLAIN',
  20.             'login_response' => null,
  21.             'locale' => 'en_US',
  22.             'connection_timeout' => 3.0,
  23.             'read_write_timeout' => 3.0,
  24.             'context' => null,
  25.             'keepalive' => false,
  26.             'heartbeat' => 0,
  27.         ],
  28.     ],
  29. ];

投递消息

使用 generator 工具新建一个 producer

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

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

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

通过container获取Producer实例，即可投递消息。以下实例直接使用ApplicationContext获取Producer其实并不合理，container具体使用请到di模块中查看。

  1. <?php
  2. use Hyperf\Amqp\Producer;
  3. use App\Amqp\Producers\DemoProducer;
  4. use Hyperf\Utils\ApplicationContext;
  6. $message = new DemoProducer(1);
  7. $producer = ApplicationContext::getContainer()->get(Producer::class);
  8. $result = $producer->produce($message);

消费消息

使用 generator 工具新建一个 consumer。

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

在DemoConsumer文件中，我们可以修改Consumer注解对应的字段来替换对应的 exchange、routingKey 和 queue。其中 $data 就是解析后的元数据。示例如下。

  1. <?php
  3. declare(strict_types=1);
  5. namespace App\Amqp\Consumers;
  7. use Hyperf\Amqp\Annotation\Consumer;
  8. use Hyperf\Amqp\Message\ConsumerMessage;
  9. use Hyperf\Amqp\Result;
  11. /**
  12.  * @Consumer(exchange="hyperf", routingKey="hyperf", queue="hyperf", nums=1)
  13.  */
  14. class DemoConsumer extends ConsumerMessage
  15. {
  16.     public function consume($data): string
  17.     {
  18.         print_r($data);
  19.         return Result::ACK;
  20.     }
  21. }

框架会根据Consumer注解自动创建Process进程，进程意外退出后会被重新拉起。