RabbitMqBundle

Instead of creating one worker per queue, you can use multiple_consumers to listen to several queues within a single consumer process. This is useful for grouping related tasks and reducing the number of active workers.

• Each queue under a multiple_consumers definition must specify its own name, callback, and routing_keys.
• The callback for each queue must implement ConsumerInterface.
• You can optionally provide a queues_provider service that implements QueuesProviderInterface to dynamically supply queues.
• All queues in a multiple_consumers block share the same exchange.

multiple_consumers:
    upload:
        connection:       default
        exchange_options: {name: 'upload', type: direct}
        queues_provider: queues_provider_service
        queues:
            upload-picture:
                name:     upload_picture
                callback: upload_picture_service
                routing_keys:
                    - picture
            upload-video:
                name:     upload_video
                callback: upload_video_service
                routing_keys:
                    - video
            upload-stats:
                name:     upload_stats
                callback: upload_stats

In Symfony 4.2+, the bundle automatically creates container aliases for producers and consumers to support dependency injection via type-hinting.

• Producers: All producers are aliased to OldSound\RabbitMqBundle\RabbitMq\ProducerInterface. It is highly recommended to use this interface for type-hinting.
• Consumers: All consumers are aliased to OldSound\RabbitMqBundle\RabbitMq\ConsumerInterface.

Argument Naming Convention: The argument name in your controller/service is constructed from the configuration key and suffixed with Producer or Consumer. For example, a configuration key upload_picture will be autowired if you use the argument name $uploadPictureProducer.

public function indexAction($name, ProducerInterface $uploadPictureProducer)
{
    $msg = array('user_id' => 1235, 'image_path' => '/path/to/new/pic.png');
    $uploadPictureProducer->publish(serialize($msg));
}

A producer sends messages to an exchange in the RabbitMQ broker. In your configuration, you define producers with connection options and exchange options (name and type).

When you define a producer named upload_picture in your configuration, the bundle provides a service in the container named old_sound_rabbit_mq.upload_picture_producer.

To send a message, use the OldSound\RabbitMqBundle\RabbitMq\Producer#publish() method. This method accepts:

1. The message body (string).
2. An optional routing key.
3. An optional array of additional properties to customize the PhpAmqpLib\Message\AMQPMessage (e.g., changing headers).

You can also explicitly set the content type or delivery mode on the producer instance using setContentType() and setDeliveryMode(). Defaults are text/plain and 2 respectively.

public function indexAction($name)
{
    $msg = array('user_id' => 1235, 'image_path' => '/path/to/new/pic.png');
    $this->get('old_sound_rabbit_mq.upload_picture_producer')->publish(serialize($msg));
}

The bundle provides built-in support for RPC patterns using Symfony services.

1. Configure the RPC Server and Client

Define rpc_servers to handle requests and rpc_clients to send them.

2. Start the RPC Server

Run the server from the CLI:

./app/console_dev rabbitmq:rpc-server <server_name>

3. Use the RPC Client in a Controller

To use a client, fetch the service named old_sound_rabbit_mq.<client_id>_rpc. Use addRequest to send a message and getReplies() to retrieve the results. getReplies() is a blocking call.

Request Parameters:

• args: The data to send (usually serialized).
• server: The name of the RPC server.
• requestId: A unique identifier for the request to find the reply in the results array.
• routingKey (optional)
• expiration (optional): Time in milliseconds before the request times out (requires RabbitMQ 3.x+).

Handling Timeouts: If a request exceeds the expiration, getReplies() will throw a \PhpAmqpLib\\Exception\\AMQPTimeoutException.

Batch consumers allow you to process a collection of messages at once (e.g., for batch database inserts).

Implementation

Your callback service must implement BatchConsumerInterface and its batchExecute(array $messages) method.

Return Values for batchExecute:

• Return true to acknowledge (ack) all messages in the batch.
• Return an array where keys are delivery tags and values indicate the action:
  • true: ack
  • false: reject and drop
  • -1: reject and requeue
  • nack and requeue: (specific logic for nack)

Configuration Example

batch_consumers:
    batch_basic_consumer:
        connection:       default
        exchange_options: {name: 'batch', type: fanout}
        queue_options:    {name: 'batch'}
        callback:         batch.basic
        qos_options:      {prefetch_size: 0, prefetch_count: 2, global: false}
        timeout_wait:     5
        auto_setup_fabric: false
        idle_timeout_exit_code: -2
        keep_alive: false
        graceful_max_execution:
            timeout: 60

Running the Consumer

Use the rabbitmq:batch:consumer command.

• Use -w to run.
• Use -b|batches <number> to stop the consumer after a specific number of batches have been consumed.

namespace AppBundle\
Service;

use OldSound\RabbitMqBundle\RabbitMq\BatchConsumerInterface;
use PhpAmqpLib\Message\AMQPMessage;

class DevckBasicConsumer implements BatchConsumerInterface
{
    /**
     * @inheritDoc
     */
    public function batchExecute(array $messages)
    {
        $result = [];
        /** @var AMQPMessage $message */
        foreach ($messages as $message) {
            $result[$message->getDeliveryTag()] = $this->executeSomeLogicPerMessage($message);
        }
        return $result;
    }
}

$ ./bin/console rabbitmq:batch:consumer batch_basic_consumer -w