NSQ客户端

NSQ 是实时的分布式消息处理平台，其设计的目的是用来大规模地处理每天数以十亿计级别的消息。 它具有分布式和去中心化拓扑结构，该结构具有无单点故障、故障容错、高可用性以及能够保证消息的可靠传递的特征。

组件要求

  • php: >=5.3.0
  • ext-json: *
  • easyswoole/easyswoole: 3.x
  • easyswoole/http-client: ^1.2.5
  • easyswoole/pool: ^1.0
  • easyswoole/spl: ^1.1
  • monolog/monolog: ~1.0
  • react/react: >=0.2.1

安装方法

composer require easyswoole/nsq

仓库地址

easyswoole/nsq

基本使用

注册Nsq服务

  1. namespace EasySwoole\EasySwoole;
  3. use App\Producer\Process as ProducerProcess;
  4. use App\Consumer\Process as ConsumerProcess;
  5. use EasySwoole\EasySwoole\Swoole\EventRegister;
  6. use EasySwoole\EasySwoole\AbstractInterface\Event;
  7. use EasySwoole\Http\Request;
  8. use EasySwoole\Http\Response;
  10. class EasySwooleEvent implements Event
  11. {
  13.     public static function initialize()
  14.     {
  15.         // TODO: Implement initialize() method.
  16.         date_default_timezone_set('Asia/Shanghai');
  17.     }
  19.     public static function mainServerCreate(EventRegister $register)
  20.     {
  21.         // TODO: Implement mainServerCreate() method.
  22.         // 生产者
  23.         \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ProducerProcess());
  24.         // 消费者
  25.         \EasySwoole\Component\Process\Manager::getInstance()->addProcess(new ConsumerProcess());
  26.     }
  28.     ......
  30. }

生产者

  1. namespace App\Producer;
  3. use EasySwoole\Component\Process\AbstractProcess;
  5. class Process extends AbstractProcess
  6. {
  7.     protected function run($arg)
  8.     {
  9.         go(function () {
  10.             $config = new \EasySwoole\Nsq\Config();
  11.             $topic  = "topic.test";
  12.             $nsqlookup = new \EasySwoole\Nsq\Lookup\Nsqlookupd($config->getNsqdUrl());
  13.             $hosts = $nsqlookup->lookupHosts($topic);
  15.             foreach ($hosts as $host) {
  16.                 $nsq = new \EasySwoole\Nsq\Nsq();
  17.                 for ($i = 0; $i < 10; $i++) {
  18.                     $msg = new \EasySwoole\Nsq\Message\Message();
  19.                     $msg->setPayload("test$i");
  20.                     $nsq->push(
  21.                         new \EasySwoole\Nsq\Connection\Producer($host, $config),
  22.                         $topic,
  23.                         $msg
  24.                     );
  25.                 }
  26.             }
  27.         });
  28.     }
  29. }

消费者

  1. namespace App\Consumer;
  3. use EasySwoole\Component\Process\AbstractProcess;
  5. class Process extends AbstractProcess
  6. {
  7.     protected function run($arg)
  8.     {
  9.         go(function () {
  10.             $topic      = "topic.test";
  11.             $config     = new \EasySwoole\Nsq\Config();
  12.             $nsqlookup  = new \EasySwoole\Nsq\Lookup\Nsqlookupd($config->getNsqdUrl());
  13.             $hosts      = $nsqlookup->lookupHosts($topic);
  14.             foreach ($hosts as $host) {
  15.                 $nsq = new \EasySwoole\Nsq\Nsq();
  16.                 $nsq->subscribe(
  17.                     new \EasySwoole\Nsq\Connection\Consumer($host, $config, $topic, 'test.consuming'),
  18.                     function ($item) {
  19.                         var_dump($item['message']);
  20.                     }
  21.                 );
  22.             }
  23.         });
  24.     }
  25. }

附赠

  1. Nsq 集群部署 docker-compose.yml 一份，使用方式如下
    1. 保证4150,4151,4160,4161,4171端口未被占用（占用后可以修改compose文件中的端口号）
    2. 根目录下，docker-compose up -d
    3. 访问localhost:4171，可以查看Web版 nsqadmin 状态。

https://github.com/easy-swoole/nsq/blob/master/docker-compose.yml