RabbitMQ-based Laravel queue driver

This package allows to use RabbitMQ queues for queued Laravel (prioritized) jobs. Fully configurable.

Installed php extension ext-amqp is required. Installation steps can be found in Dockerfile.

For jobs delaying you also should install rabbitmq-delayed-message-exchange plugin for RabbitMQ. Delaying is optional feature.

Install

Important: Before using this package you should install avto-dev/amqp-rabbit-manager into your application. Installation steps can be found here.

Require this package with composer using the following command:

$ composer require avto-dev/amqp-rabbit-laravel-queue "^2.0"

Installed composer is required (how to install composer). Also you need to fix the major version of package.

You need to fix the major version of package.

After that you should modify your configuration files:

`./config/rabbitmq.php`

RabbitMQ queues and exchanges configuration:

<?php

use Interop\Amqp\AmqpQueue;
use Interop\Amqp\AmqpTopic;

return [

    // ...

    'queues' => [

        'jobs' => [
            'name'         => env('JOBS_QUEUE_NAME', 'jobs'),
            'flags'        => AmqpQueue::FLAG_DURABLE, // Remain queue active when a server restarts
            'arguments'    => [
                'x-max-priority' => 255, // @link <https://www.rabbitmq.com/priority.html>
            ],
            'consumer_tag' => null,
        ],

        'failed' => [
            'name'         => env('FAILED_JOBS_QUEUE_NAME', 'failed-jobs'),
            'flags'        => AmqpQueue::FLAG_DURABLE,
            'arguments'    => [
                'x-message-ttl' => 604800000, // 7 days, @link <https://www.rabbitmq.com/ttl.html>
                'x-queue-mode'  => 'lazy', // @link <https://www.rabbitmq.com/lazy-queues.html>
            ],
            'consumer_tag' => null,
        ],

    ],

    // ...

    'exchanges' => [

        // RabbitMQ Delayed Message Plugin is required (@link: <https://git.io/fj4SE>)
        'delayed-jobs' => [
            'name'      => env('DELAYED_JOBS_EXCHANGE_NAME', 'jobs.delayed'),
            'type'      => 'x-delayed-message',
            'flags'     => AmqpTopic::FLAG_DURABLE, // Remain active when a server restarts
            'arguments' => [
                'x-delayed-type' => AmqpTopic::TYPE_DIRECT,
            ],
        ],

    ],

    // ...

    'setup' => [
        'rabbit-default' => [
            'queues' => [
                'jobs',
                'failed',
            ],
            'exchanges' => [
                'delayed-jobs'
            ],
        ],
    ],
];

`./config/queue.php`

Laravel queue settings:

<?php

use AvtoDev\AmqpRabbitLaravelQueue\Connector;

return [

    // ...

    'default' => env('QUEUE_DRIVER', 'rabbitmq'),

    // ...

    'connections' => [

        // ...

        'rabbitmq' => [
            'driver'              => Connector::NAME,
            'connection'          => 'rabbit-default',
            'queue_id'            => 'jobs',
            'delayed_exchange_id' => 'delayed-jobs',
            'timeout'             => (int) env('QUEUE_TIMEOUT', 0), // The timeout is in milliseconds
            'resume'              => (bool) env('QUEUE_RESUME', false), // Resume consuming when timeout is over
        ],
    ],

    // ...

    'failed' => [
        'connection' => 'rabbit-default',
        'queue_id'   => 'failed',
    ],
];

resume can be used with non-zero timeout value for periodic connection reloading (for example, if you set 'timeout' => 30000 and 'resume' => true, queue worker will unsubscribe and subscribe back to the queue every 30 seconds without process exiting).

You can remove delayed_exchange_id for disabling delayed jobs feature.

At the end, don't forget to execute command php ./artisan rabbit:setup.

How jobs delaying works?

Very simple:

Usage

You can dispatch your jobs as usual (dispatch(new Job) or dispatch(new Job)->delay(10)), commands like queue:work, queue:failed, queue:retry and others works fine.

Additional features:

Jobs delaying (plugin rabbitmq_delayed_message_exchange for RabbitMQ server is required);
Jobs priority (job should implements PrioritizedJobInterface interface);
Automatically delayed messages exchanges bindings (only if you use command rabbit:setup for queues and exchanges creation);
The ability to store the state of job

State storing

Using this package you can store any valiables (except resources and callable entities) between job restarts (just use trait WithJobStateTrait in your job class). But you should remember - state is available only inside job handle method.

⚠️ Warning

Be careful with commands queue:failed and queue:retry. If during command execution something happens (lost connection, etc) you may loose all failed jobs!

You should avoid to use next method (broker does not guarantee operations order, so calling results may be wrong):

\AvtoDev\AmqpRabbitLaravelQueue\Queue::size()
\AvtoDev\AmqpRabbitLaravelQueue\Failed\RabbitQueueFailedJobProvider::count()

Testing

For package testing we use phpunit framework and docker-ce + docker-compose as develop environment. So, just write into your terminal after repository cloning:

$ make build
$ make latest # or 'make lowest'
$ make test

Changes log

Changes log can be found here.

Support

If you will find any package errors, please, make an issue in current repository.

License

This is open-sourced software licensed under the MIT License.

eldario/amqp-rabbit-laravel-queue