dudashuang/php-queue

a php client for message queue

php-queue

, _(*1)

A php client for message queue which is one of RabbitMQ, Kafka and Redis., _(*2)

Requirement

• redis, _(*3)

  sudo apt-get install redis-server
  

• RabbitMQ, _(*4)

  http://dudashuang.com/rabbitmq/, _(*5)

• Kafka, _(*6)

  http://dudashuang.com/kafka/, _(*7)

Install

• composer, _(*8)

  composer require dudashuang/php-queue
  

Base Usage

• examples, _(*9)

  • job
  • event
  • listener
  • ListenerServiceProvider

• create a driver, _(*10)

  • redis:

  <?php
  require __DIR__ . '/vendor/autoload.php';
  
  $connector  = new Lily\Connectors\RedisConnector([
      'scheme'             => 'tcp',
      'host'               => '127.0.0.1',
      'port'               => 6379,
      'read_write_timeout' => 0,
  ]);
  $driver = new Lily\Drivers\Redis($connector);
  

  • kafka:

  <?php
  require __DIR__ . '/vendor/autoload.php';
  
  $connector  = new Lily\Connectors\KafkaConnector([
      'brokers' => [
          ['host' => 'localhost', 'port' => 9092],
      ],
  ]);
  $driver = new Lily\Drivers\Kafka($connector);
  

  • rabbitmq:

  <?php
  require __DIR__ . '/vendor/autoload.php';
  
  $connector  = new Lily\Connectors\RabbitMQConnector([
      'host'     => 'localhost',
      'port'     => 5672,
      'username' => 'guest',
      'password' => 'guest',
  ]);
  $driver = new Lily\Drivers\RabbitMQ($connector);
  

• create a application, _(*11)

  <?php
  require __DIR__ . '/vendor/autoload.php';
  
  $application = new Lily\Application($driver, [
      'deafult_queue' => 'default-queue',
      'failed_queue'  => 'failed-queue',
  ]);
  

• dispatch a job, _(*12)

  • default queue

  for ($i=0; $i<10; $i++) {
      $application->dispatch(new TestJob('hello', new LilyTest\TestModel(1, 2)));
  }
  

  • other queue

  $application->dispatch((new TestJob(...))->set_queue($queue_name));
  

• dispatch a event, _(*13)

  $application->dispatch(new TestEvent(...));
  

• create a consumer, _(*14)

  $application−>consume($queue_name);
  

• create a listener, _(*15)

  $application->listen('LilyTest\Listeners\SendListener', ['TestEvent', 'TestEvent1']);
  

TODO

• add RocketMQ driver, _(*16)

• add delay queue, _(*17)

Additional

you can use supervisor to manage consumer, _(*18)

• install, _(*19)

  sudo apt-get install supervisor
  

• supervisor config, _(*20)

  sudo vim /etc/supervisor/conf.d/guopika.config
  

  [program:guopika-worker]
  process_name=%(program_name)s_%(process_num)02d
  command=php /path_to_listen_queue/cli_func
  autostart=true
  autorestart=true
  user=www-data
  numprocs=4
  redirect_stderr=true
  stdout_logfile=/path/worker.log