Быстрый старт работы с RabbitMq

---

Создаем канал

Самое первое, что необходимо сделать, это реализовать сервис, который будет работать с конкретным каналом очередей.

1. Создаем файл src/AppBundle/Service/jobChannelTest.php [посмотреть код]

   jobChannelTest.php

   ×

   <?php
   
   namespace AppBundle\Service;
   
   use Saq\RabbitMqQueueBundle\RabbitMq\MqChannel;
   use Saq\RabbitMqQueueBundle\RabbitMq\MqJob;
   use Symfony\Component\DependencyInjection\ContainerInterface;
   
   /**
    * class:  jobChannelTest
    * file:  src/AppBundle/Service/jobChannelTest.php
    */
   class jobChannelTest extends MqChannel
   {
   
   	/**
   	 * Имя индефикатора канала очередей
   	 * @var string
   	 */
   	public $queue = 'testQueue';
   
   	public function __construct(ContainerInterface $container)
   	{
   		parent::__construct($container);
   	}
   
   	/**
   	 * Метод, в который получаем текущее задание из очереди
   	 * @param MqJob $job
   	 * @return mixed|void
   	 */
   	public function processJob(MqJob $job)
   	{
   		$val = $job->getData();
   
   		if ($val == 'jobdelete') {
   			// например задание успешно выполнено, мы его удалем из очереди
   			$job->delete();
   
   		} elseif ($val == 'jobskip') {
   			// например задание НЕ выполнено, мы его пропускаем и переходим к обработке следующего.
   			// Само задание остается в очереди, и при следующем вызове слушателя опять попадет в обработку
   			$job->skip();
   
   		} elseif ($val == 'jobcancel') {
   			// Если надо остановить всю очередь разработки, при этом текущее задание останется в очереди
   			$job->cancel();
   		}
   	}
   }

2. Регистрируем его как сервис (services.yml)

   # ....
       app.job.channel.test:
           class: AppBundle\Service\jobChannelTest
           public: true

Помещаем в нашу очередь задания

• Получаем сервис управления очередью через контейнер

  $channelTest = $this->get('app.job.channel.test');

• помещаем в очередь несколько заданий

  		$channelTest
  			->addJob(['key' => 'val'])   // перед сохранением массив будет австоматически конвертирован в JSON строку
  			->addJob('jobdelete')  // использем строку как кей для удлаения задания при обработке
  			->addJob('jobskip')    // использем строку как кей для пропуска задания при обработке
  			->addJob('jobcancel')  // использем строку как кей для остановки обработки
  		;

Реализуем чтение очереди

• Создаем файл src/AppBundle/Command/listenerJobsTestCommand.php [посмотреть код]

  Это у нас консольная команда, которую можно выполнить как разово, так и поставить на постоянное выполнение в CRON с периодичностью запуска в минуту

  jobChannelTest.php

  ×

  <?php
  
  namespace AppBundle\Command;
  
  use Saq\RabbitMqQueueBundle\Command\MqJobsListener;
  
  /**
   * class:  listenerJobsTestCommand
   * file:  src/AppBundle/Command/listenerJobsTestCommand.php
   */
  class listenerJobsTestCommand extends MqJobsListener
  {
  	/**
  	 * php bin/console mq.jobs:listener.test
  	 */
  	protected $commandName = 'mq.jobs:listener.test';
  
  	/**
  	 * @var string
  	 */
  	protected $commandDescription = 'Листенер очереди канала тест';
  
  	/**
  	 * Количество запущенных копий демона
  	 * @var int
  	 */
  	protected $countProcesses = 2;
  
  	protected function runCommand()
  	{
  		$this->getContainer()->get('app.job.channel.test')->listen();
  	}
  }

• Выполняем команду в консоли. Если все сделали верно, то ошибок не возникнет

  php bin/console mq.jobs:listener.test