NATS

NATS 是一个开源、轻量级、高性能的分布式消息中间件，实现了高可伸缩性和优雅的 Publish / Subscribe 模型，使用 Golang 语言开发。NATS 的开发哲学认为高质量的 QoS 应该在客户端构建，故只建立了 Request-Reply，不提供 1. 持久化 2. 事务处理 3. 增强的交付模式 4. 企业级队列。

安装

  1. composer require hyperf/nats

使用

创建消费者

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

如果设置了 queue，则相同的 subject 只会被一个 queue 消费。若不设置 queue，则每个消费者都会受到消息。

  1. <?php
  3. declare(strict_types=1);
  5. namespace App\Nats\Consumer;
  7. use Hyperf\Nats\AbstractConsumer;
  8. use Hyperf\Nats\Annotation\Consumer;
  9. use Hyperf\Nats\Message;
  11. #[Consumer(subject: 'hyperf.demo', queue: 'hyperf.demo', name: 'DemoConsumer', nums: 1)]
  12. class DemoConsumer extends AbstractConsumer
  13. {
  14.     public function consume(Message $payload)
  15.     {
  16.         // Do something...
  17.     }
  18. }

投递消息

使用 publish 投递消息。

  1. <?php
  3. declare(strict_types=1);
  5. namespace App\Controller;
  7. use Hyperf\Di\Annotation\Inject;
  8. use Hyperf\HttpServer\Annotation\AutoController;
  9. use Hyperf\Nats\Driver\DriverInterface;
  11. #[AutoController(prefix: "nats")]
  12. class NatsController extends AbstractController
  13. {
  14.     #[Inject]
  15.     protected DriverInterface $nats;
  17.     public function publish()
  18.     {
  19.         $res = $this->nats->publish('hyperf.demo', [
  20.             'id' => 'Hyperf',
  21.         ]);
  23.         return $this->response->success($res);
  24.     }
  25. }

使用 request 投递消息。

  1. <?php
  3. declare(strict_types=1);
  5. namespace App\Controller;
  7. use Hyperf\Di\Annotation\Inject;
  8. use Hyperf\HttpServer\Annotation\AutoController;
  9. use Hyperf\Nats\Driver\DriverInterface;
  10. use Hyperf\Nats\Message;
  12. #[AutoController(prefix: "nats")]
  13. class NatsController extends AbstractController
  14. {
  15.     #[Inject]
  16.     protected DriverInterface $nats;
  18.     public function request()
  19.     {
  20.         $res = $this->nats->request('hyperf.reply', [
  21.             'id' => 'limx',
  22.         ], function (Message $payload) {
  23.             var_dump($payload->getBody());
  24.         });
  26.         return $this->response->success($res);
  27.     }
  28. }

使用 requestSync 投递消息。

  1. <?php
  3. declare(strict_types=1);
  5. namespace App\Controller;
  7. use Hyperf\Di\Annotation\Inject;
  8. use Hyperf\HttpServer\Annotation\AutoController;
  9. use Hyperf\Nats\Driver\DriverInterface;
  10. use Hyperf\Nats\Message;
  12. #[AutoController(prefix: "nats")]
  13. class NatsController extends AbstractController
  14. {
  15.     #[Inject]
  16.     protected DriverInterface $nats;
  18.     public function sync()
  19.     {
  20.         /** @var Message $message */
  21.         $message = $this->nats->requestSync('hyperf.reply', [
  22.             'id' => 'limx',
  23.         ]);
  25.         return $this->response->success($message->getBody());
  26.     }
  27. }