NATS

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

使用

创建消费者

  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. /**
  12.  * @Consumer(subject="hyperf.demo", queue="hyperf.demo", name="DemoConsumer", nums=1)
  13.  */
  14. class DemoConsumer extends AbstractConsumer
  15. {
  16.     public function consume(Message $payload)
  17.     {
  18.         // Do something...
  19.     }
  20. }

投递消息

使用 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. /**
  12.  * @AutoController(prefix="nats")
  13.  */
  14. class NatsController extends Controller
  15. {
  16.     /**
  17.      * @Inject
  18.      * @var DriverInterface
  19.      */
  20.     protected $nats;
  22.     public function publish()
  23.     {
  24.         $res = $this->nats->publish('hyperf.demo', [
  25.             'id' => 'Hyperf',
  26.         ]);
  28.         return $this->response->success($res);
  29.     }
  30. }

使用 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. /**
  13.  * @AutoController(prefix="nats")
  14.  */
  15. class NatsController extends Controller
  16. {
  17.     /**
  18.      * @Inject
  19.      * @var DriverInterface
  20.      */
  21.     protected $nats;
  23.     public function request()
  24.     {
  25.         $res = $this->nats->request('hyperf.reply', [
  26.             'id' => 'limx',
  27.         ], function (Message $payload) {
  28.             var_dump($payload->getBody());
  29.         });
  31.         return $this->response->success($res);
  32.     }
  33. }

使用 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. /**
  13.  * @AutoController(prefix="nats")
  14.  */
  15. class NatsController extends Controller
  16. {
  17.     /**
  18.      * @Inject
  19.      * @var DriverInterface
  20.      */
  21.     protected $nats;
  23.     public function sync()
  24.     {
  25.         /** @var Message $message */
  26.         $message = $this->nats->requestSync('hyperf.reply', [
  27.             'id' => 'limx',
  28.         ]);
  30.         return $this->response->success($message->getBody());
  31.     }
  32. }