Redis adapater for the symfony/messenger component

1.0.9 2018-05-18 15:48 UTC


Build Status

This is an experimental Receiver/Sender on Redis for the symfony/messenger component.

Quick start

For now we're exposing a bundle which is pre-configuring the Messenger component with receivers and senders.

composer require symfony/messenger soyuka/symfony-messenger-redis

Add the bundle new Soyuka\RedisMessengerAdapter\Bundle\RedisMessengerAdapterBundle().

Also requires the redis extension.

Add the following configuration:

        'App\Message\Foo': 'foo_queue'

Add a message handler:


namespace App\MessageHandler;

use App\Message\Foo;

final class FooHandler
    public function __invoke(Foo $message)

Tag it:

          - { name: messenger.message_handler }

You're done!

Launch bin/console messenger:consume-messages redis_messenger.receiver.foo_queue and dispatch messages from the bus:

$bus->dispatch(new Foo());

Configuration reference

        url: ''
        port: 6379
        serializer: !php/const \Redis::SERIALIZER_IGBINARY # default is \Redis::SERIALIZER_PHP
        'App\Message\Foo': 'foo_queue'
            queue: 'bar_queue'
            ttl: 10000
            blockingTimeout: 1000


Relevant discussion:

The sender uses a List and uses RPUSH (add value to the tail of the list). The receiver uses BRPOPLPUSH which reads the last element of the list and adds in to the head of another list (queue_processing). If no elements are present it'll block the connection until a new element shows up or the timeout is reached. When timeout is reached it works like a "ping" of some sort (todo wait for 26632 to be merged and $handle(null)). On every iteration, we will check the queue_processing list. For every items in this queue we have a corresponding key in redis with a given ttl. If the key has expired, the item is LREM (removed) from queue_processing and put back in the origin queue to be processed again. This workaround helps avoiding lost messages.

I started a RedisAdapter and may add it to symfony once messenger documentation and AMQP adapter are merged.