edipoelwes / laravel-rabbitmq-worker
This project is a Laravel package that facilitates the execution of RabbitMQ worker processes. It simplifies the configuration, management, and monitoring of worker processes, allowing you to easily integrate RabbitMQ into your Laravel application to handle asynchronous tasks efficiently and scalabl
Package info
github.com/edipoelwes/laravel-rabbitmq-worker
pkg:composer/edipoelwes/laravel-rabbitmq-worker
Requires
- php: ^7.4|^8.0
- illuminate/console: ^6.0|^7.0|^8.0|^9.0|^10.0|^11.0
- illuminate/support: ^6.0|^7.0|^8.0|^9.0|^10.0|^11.0
- php-amqplib/php-amqplib: ^3.6
This package is auto-updated.
Last update: 2026-07-14 18:57:53 UTC
README
Library to facilitate the use of rabbitmq within php based on the php-amqplib library.
Installing
composer require edipoelwes/laravel-rabbitmq-worker
How to configure in Laravel
Run the publisher to generate configuration file
When you run the publisher command, it automatically creates a configuration file named laravel-rabbitmq-worker.php within the config directory of your Laravel application.
php artisan vendor:publish --provider="Edipoelwes\LaravelRabbitmqWorker\CommandServiceProvider"
Clear settings cache
Before configuring the RabbitMQ connection settings according to your environment, it's essential to clear the Laravel configuration cache to ensure that any changes take effect properly.
php artisan config:cache
Then just configure according to your environment.
<?php return [ 'connections' => [ 'host' => env('RABBITMQ_HOST', 'localhost'), 'port' => env('RABBITMQ_PORT', 5672), 'user' => env('RABBITMQ_LOGIN', 'guest'), 'password' => env('RABBITMQ_PASSWORD', 'guest'), 'vhost' => env('RABBITMQ_VHOST', '/'), ] ];
Usage examples
Creating a simple publisher
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'queue', // Queue 'route-key', // Routing key '', // Exchange '', // Exchange Type '', // Consumer Tag false, // Passive true, // Durable false, // Exclusive false // Auto delete ); // Prepare response payload $payload = "your message"; $rabbitMQService->publish($payload); $rabbitMQService->destruct(); // Clean up resources
Publishing with message headers
Use message headers when your consumer routing depends on AMQP application_headers. This is separate from queue declaration arguments.
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'dasa_priority_low', 'dasa_priority_low' ); $rabbitMQService->publishWithHeaders( json_encode(['fiad_id' => 123]), ['message_type' => 'dasa_update_pbzpa'] ); $rabbitMQService->destruct();
Publishing a batch with message headers
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'dasa_priority_low', 'dasa_priority_low' ); $rabbitMQService->publishBatchWithHeaders( [ ['fiad_id' => 1], ['fiad_id' => 2], ], ['message_type' => 'dasa_update_pbzpa'] ); $rabbitMQService->destruct();
Creating a consumer
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'queue', // Queue 'route-key', // Routing key '', // Exchange '', // Exchange Type '', // Consumer Tag false, // Passive true, // Durable false, // Exclusive false, // Auto delete [] // Custom arguments (optional) ); $callback = function ($msg) { // $msg->body // your code here $msg->ack(); } $rabbitMQService->consume($callback); $rabbitMQService->destruct(); // Clean up resources
Creating a Consumer Using QueueConsumerAbstract
You can create a consumer by extending the QueueConsumerAbstract class. This approach allows you to handle messages in a structured way. Below is an example of how to implement a command that consumes messages from a RabbitMQ queue:
<?php namespace App\Console\Commands; use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\QueueConsumerAbstract; use Illuminate\Support\Facades\Log; class RabbitmqTesting extends QueueConsumerAbstract { protected $signature = 'hello-command'; protected string $queueName = 'hello'; protected string $routeKey = 'hello'; protected string $consumerTag = 'rabbitmq-command-testing'; // Optional: override $arguments to set custom queue configuration (TTL, DLQ, delivery limit, etc.) // See "Custom Queue Arguments" section for examples. /** * @throws \Exception */ public function process($message): void { Log::info('start handling'); Log::info($message->body); $message->ack(); Log::info('finish handling'); } }
Creating a consumer into laravel command
<?php namespace App\Console\Commands; use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; use Illuminate\Console\Command; class Consume extends Command { /** * The name and signature of the console command. * * @var string */ protected $signature = 'your-command'; /** * The console command description. * * @var string */ protected $description = 'Command description'; /** * Create a new command instance. * * @return void */ public function __construct() { parent::__construct(); } /** * Execute the console command. * * @return int */ public function handle() { $rabbitMQService = new RabbitMQService( 'queue', // Queue 'route-key', // Routing key '', // Exchange '', // Exchange Type '', // Consumer Tag false, // Passive true, // Durable false, // Exclusive false, // Auto delete [] // Custom arguments (optional) ); $callback = function ($msg) { // $msg->body // your code here $msg->ack(); } $rabbitMQService->consume($callback); $rabbitMQService->destruct(); // Clean up resources } }
Custom Queue Arguments
You can pass custom RabbitMQ queue arguments to the consumer via the $arguments parameter in RabbitMQService or through QueueBuilder::setArguments(). This is useful for features like message TTL, dead-letter exchanges, delivery limits, and more.
Each argument follows the AMQP type notation: 'argument-name' => ['type', value], where the type is 'S' for string and 'I' for integer.
setArguments()replaces the arguments array entirely — what you pass is exactly what reaches the broker (merged with the base defaultx-queue-type: quorumin the constructor). To override the default, includex-queue-typein your own arguments.
Quorum replication (x-quorum-initial-group-size)
Whenever the final queue type is quorum, the lib also sends
x-quorum-initial-group-size so the queue is born replicated across the
expected number of cluster nodes. The value comes from
config('laravel-rabbitmq-worker.quorum.initial_group_size'):
RABBITMQ_QUORUM_INITIAL_GROUP_SIZE=3 # default: 3 (set 0 to omit the argument)
Notes:
- The argument only takes effect when the queue is created. Queues that
already exist on the broker keep their current members — grow them with
rabbitmq-queues grow <node> allor delete/redeclare the queue. - With 3 replicas a quorum queue tolerates 1 node down (majority of 2); a queue created with a single member becomes unavailable whenever its node goes down.
- Passing
x-quorum-initial-group-sizeexplicitly in your own arguments takes precedence over the config value.
Using QueueConsumerAbstract, override the $arguments property in your subclass:
<?php namespace App\Console\Commands; use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\QueueConsumerAbstract; class RabbitmqTesting extends QueueConsumerAbstract { protected $signature = 'hello-command'; protected string $queueName = 'hello'; protected string $routeKey = 'hello'; protected string $consumerTag = 'rabbitmq-command-testing'; protected array $arguments = [ 'x-message-ttl' => ['I', 15000], // TTL in ms (integer) 'x-dead-letter-exchange' => ['S', ''], // dead-letter exchange (string) 'x-dead-letter-routing-key' => ['S', 'your-queue-name.dlq'], // DLQ routing key (string) 'x-delivery-limit' => ['I', 3], // retry 3x, on 4th goes to DLQ ]; public function process($message): void { // your code here $message->ack(); } }
Using RabbitMQService directly:
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'queue', 'route-key', '', '', '', false, true, false, false, [ 'x-message-ttl' => ['I', 15000], // TTL in ms (integer) 'x-dead-letter-exchange' => ['S', ''], // dead-letter exchange (string) 'x-dead-letter-routing-key' => ['S', 'your-queue-name.dlq'], // DLQ routing key (string) 'x-delivery-limit' => ['I', 3], // retry 3x, on 4th goes to DLQ ] ); $rabbitMQService->consume(function ($msg) { // your code here $msg->ack(); }); $rabbitMQService->destruct();
Using QueueBuilder:
<?php use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\QueueBuilder; $queue = (new QueueBuilder()) ->setQueueName('queue') ->setRouteKey('route-key') ->setIsDurable(true) ->setArguments([ 'x-message-ttl' => ['I', 15000], 'x-dead-letter-exchange' => ['S', ''], 'x-dead-letter-routing-key' => ['S', 'your-queue-name.dlq'], 'x-delivery-limit' => ['I', 3], ]) ->getQueue(); $queue->consume(function ($msg) { // your code here $msg->ack(); }); $queue->destruct();
Using QueueProducer:
<?php use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\QueueProducer; // Inject QueueProducer via constructor or resolve from container $queueProducer->produce('your-queue-name', $payload, [ 'x-message-ttl' => ['I', 15000], 'x-dead-letter-exchange' => ['S', ''], 'x-dead-letter-routing-key' => ['S', 'your-queue-name.dlq'], 'x-delivery-limit' => ['I', 3], ]); // Batch publish $queueProducer->produceBatch('your-queue-name', $messages, [ 'x-message-ttl' => ['I', 15000], 'x-dead-letter-exchange' => ['S', ''], 'x-dead-letter-routing-key' => ['S', 'your-queue-name.dlq'], 'x-delivery-limit' => ['I', 3], ]); // Publish with AMQP application headers $queueProducer->produceWithHeaders( 'dasa_priority_low', ['fiad_id' => 123], ['message_type' => 'dasa_update_pbzpa'] ); // Batch publish with AMQP application headers $queueProducer->produceBatchWithHeaders( 'dasa_priority_low', [ ['fiad_id' => 1], ['fiad_id' => 2], ], ['message_type' => 'dasa_update_pbzpa'] );
Note: The
$argumentsparameter inQueueProduceris optional. When provided, they must match the arguments used by the consumer to avoidPRECONDITION_FAILEDerrors.
argumentsconfigure the queue itself (x-message-ttl, DLQ, quorum, etc.).headersconfigure the message being published (message_type, tenant, trace metadata, etc.). They solve different problems and should not be mixed.
Remote procedure call (RPC)
Configuration for RPC
For publishing and consuming with RPC, please ensure that the RabbitMQ class is configured with 'durable = false' and 'auto_delete = true' settings.
When setting up Remote Procedure Call (RPC) functionality, it's crucial to configure RabbitMQ appropriately to ensure seamless communication. For optimal performance and resource management, it's recommended to adjust the RabbitMQ settings with 'durable = false' and 'auto_delete = true'.
Creating a Consumer for RPC
<?php use Edipoelwes\LaravelRabbitmqWorker\Services\RabbitMQ\RabbitMQService; $rabbitMQService = new RabbitMQService( 'rpc_queue', // Queue 'rpc_queue', // Routing key '', // Exchange '', // Exchange Type 'rpc_queue', // Consumer Tag false, // Passive false, // Durable false, // Exclusive true // Auto delete ); // Define RPC Consumer Callback $rabbitMQService->consume(function ($req) { // Your code here // Prepare response payload $payload = "your message"; // Publish response to the specified channel and correlation ID $msg = new AMQPMessage( $payload, array('correlation_id' => $req->get('correlation_id')) ); $req->getChannel()->basic_publish( $msg, '', $req->get('reply_to') ); $req->ack(); // Acknowledge message processing }); $rabbitMQService->destruct(); // Clean up resources
Creating a Publisher for RPC
<?php // Configure RabbitMQ for RPC Publisher $rabbitMQService = new RabbitMQService( 'rpc_queue', // Queue 'rpc_queue', // Routing key '', // Exchange '', // Exchange Type 'rpc_queue', // Consumer Tag false, // Passive false, // Durable false, // Exclusive true // Auto delete ); // Prepare response payload $payload = "your message"; $response = $rabbitMQService->publishRpc($payload); // your code here $rabbitMQService->destruct(); // Clean up resources
Priority queues (high/default/low)
Em vez de uma fila dedicada por ação, consolide o consumo em 3 filas físicas por
prioridade. Cada mensagem carrega o header AMQP message_type, e o
PriorityMessageRouter resolve o consumer da aplicação em
config('laravel-rabbitmq-worker.priority.routes').
1. Configure prefixo, filas e rotas (config publicado da aplicação)
// config/laravel-rabbitmq-worker.php // ATENÇÃO: o mergeConfigFrom é raso — defina o bloco `priority` completo. 'command_prefix' => env('RABBITMQ_COMMAND_PREFIX', 'meuapp'), 'priority' => [ 'queues' => [ 'high' => env('RABBITMQ_PRIORITY_QUEUE_HIGH', 'meuapp_priority_high'), 'default' => env('RABBITMQ_PRIORITY_QUEUE_DEFAULT', 'meuapp_priority_default'), 'low' => env('RABBITMQ_PRIORITY_QUEUE_LOW', 'meuapp_priority_low'), ], 'dead_letter' => [ 'enabled' => env('RABBITMQ_PRIORITY_DLQ_ENABLED', true), 'suffix' => env('RABBITMQ_PRIORITY_DLQ_SUFFIX', '.dlq'), 'delivery_limit' => (int) env('RABBITMQ_PRIORITY_DELIVERY_LIMIT', 3), 'queue_type' => env('RABBITMQ_PRIORITY_DLQ_QUEUE_TYPE', 'quorum'), 'priorities' => [ 'high' => [], 'default' => [], 'low' => [], ], ], 'routes' => [ 'meu_message_type' => [ 'priority' => 'default', 'consumer' => \App\Consumers\MeuConsumer::class, // classe com process($message) ], ], ],
E no .env:
RABBITMQ_COMMAND_PREFIX=meuapp
2. Os workers de fila já vêm prontos na lib
A lib registra nativamente 3 commands Artisan de consumo — um por prioridade —
usando PriorityQueueConsumerAbstract internamente. O nome final de cada
command respeita o command_prefix configurado:
<command_prefix>_priority_high<command_prefix>_priority_default<command_prefix>_priority_low
Com RABBITMQ_COMMAND_PREFIX=meuapp, isso vira meuapp_priority_high,
meuapp_priority_default e meuapp_priority_low — nomes que você usa
diretamente em RABBITMQ_LARAVEL_WORKERS / php artisan rabbitmq:run. Não é
necessário criar nenhuma subclasse na aplicação para isso.
Se precisar de um worker com nome fora desse padrão (caso raro), ainda é
possível estender PriorityQueueConsumerAbstract diretamente e definir
$signature manualmente:
use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\PriorityQueueConsumerAbstract; class PriorityDefaultCommand extends PriorityQueueConsumerAbstract { protected $signature = 'nome-fora-do-padrao'; protected string $priority = 'default'; }
3. Crie o consumer da ação (PriorityConsumerInterface)
use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\PriorityConsumerInterface; class MeuConsumer implements PriorityConsumerInterface { public function process($message): void { $data = json_decode($message->body, true); // regra de negócio; o consumer decide ack/nack/reject $message->ack(); } }
4. Publique com o header de roteamento
Forma recomendada — a prioridade é inferida automaticamente de
priority.routes.<message_type>.priority, então você não repete
'high'|'default'|'low' no código da aplicação:
use Edipoelwes\LaravelRabbitmqWorker\Infrastructure\Queue\Rabbitmq\QueueProducer; app(QueueProducer::class)->produceRouted('meu_message_type', ['id' => 123]); app(QueueProducer::class)->produceRoutedBatch('meu_message_type', $payloads);
produceRouted()/produceRoutedBatch() lançam InvalidArgumentException se
meu_message_type não estiver mapeado em priority.routes (ou a rota não
definir priority) — isso é intencional: preferimos falhar rápido a publicar
silenciosamente em default.
producePriority()/producePriorityBatch() continuam existindo e funcionando
exatamente como antes — use-as quando precisar de um override explícito da
prioridade (ex.: um job que decide a prioridade em tempo de execução, fora do
que está mapeado em priority.routes):
app(QueueProducer::class)->producePriority('low', 'meu_message_type', ['id' => 123]); app(QueueProducer::class)->producePriorityBatch('low', 'meu_message_type', $payloads);
4.1. Publicando no namespace de outro sistema (remotes)
Quando dois sistemas compartilham o mesmo broker e cada um é dono das suas
filas de prioridade (ex.: dasa_priority_* consumidas pelo DASA e
eco-utm_priority_* consumidas pelo Eco UTM), o produtor publica nas filas do
outro sistema declarando a topologia dele em priority.remotes:
'priority' => [ // topologia LOCAL: as filas que ESTE app consome 'queues' => [ 'high' => 'eco-utm_priority_high', 'default' => 'eco-utm_priority_default', 'low' => 'eco-utm_priority_low', ], // topologias REMOTAS: filas de outros sistemas, só para publicação. // queues e dead_letter DEVEM espelhar a config do sistema remoto, // senão o queue_declare do produtor gera PRECONDITION_FAILED. 'remotes' => [ 'dasa' => [ 'queues' => [ 'high' => 'dasa_priority_high', 'default' => 'dasa_priority_default', 'low' => 'dasa_priority_low', ], 'dead_letter' => [ 'enabled' => true, 'suffix' => '.dlq', 'delivery_limit' => 3, 'queue_type' => 'quorum', ], ], ], 'routes' => [ // outbound: publica na fila do DASA; sem 'consumer' (quem processa é o DASA) 'eco-utm_flight-requests' => ['priority' => 'high', 'remote' => 'dasa'], // inbound/interno: fila local, com consumer deste app 'minha_rotina_interna' => ['priority' => 'low', 'consumer' => MinhaRotina::class], ], ],
produceRouted()/produceRoutedBatch() inferem o remote da rota — o código
da aplicação não muda. Para override explícito, producePriority() e
producePriorityBatch() aceitam o namespace como quinto parâmetro:
app(QueueProducer::class)->produceRouted('eco-utm_flight-requests', ['id' => 123]); // vai para dasa_priority_high app(QueueProducer::class)->producePriority('high', 'meu_message_type', $payload, [], 'dasa');
O consumo é sempre local: os commands <prefix>_priority_* só consomem
priority.queues, e a criação/consumo das DLQs de um namespace remoto é
responsabilidade do sistema dono das filas. Uma rota com remote aponta para
um nome inexistente em priority.remotes? InvalidArgumentException — falha
rápido em vez de publicar na fila errada.
5. Teste manual
rabbitmq:priority-publish usa a mesma inferência por padrão. Passe
--priority só quando quiser sobrescrever a rota (ou testar um message_type
ainda não mapeado):
php artisan rabbitmq:priority-publish meu_message_type --payload='{"id":123}'
php artisan rabbitmq:priority-publish meu_message_type --priority=low --count=5
Mensagens sem header message_type ou com tipo não mapeado são rejeitadas sem
requeue (reject(false)) e logadas — a fila nunca trava por mensagem
desconhecida. Lembre-se: workers são processos de longa duração; após alterar
priority.routes, reinicie os workers.
6. DLQ por fila de prioridade (evita loop infinito)
Cada fila física de prioridade (priority_high, priority_default,
priority_low) ganha automaticamente uma DLQ irmã (<fila>.dlq por padrão).
A DLQ é por fila de prioridade, não por message_type — isso mantém o
modelo simples (3 filas + 3 DLQs) mesmo que o app tenha dezenas de rotas em
priority.routes.
Isso é resolvido de forma centralizada por PriorityQueueTopology e usado
tanto por PriorityQueueConsumerAbstract (declara a fila principal com os
argumentos de DLQ e garante que a .dlq exista) quanto por
QueueProducer::producePriority()/producePriorityBatch() — e por extensão
produceRouted()/produceRoutedBatch(), que delegam para elas — aplicando os
mesmos argumentos ao publicar, evitando PRECONDITION_FAILED por divergência
de argumentos entre quem publica e quem consome.
Com dead_letter.enabled (padrão true), a fila principal é declarada com:
x-dead-letter-exchange = ''
x-dead-letter-routing-key = '<fila>.dlq'
x-delivery-limit = dead_letter.delivery_limit (padrão 3)
Fluxo resultante dentro do process($message) do seu consumer:
$message->ack()— mensagem concluída, some da fila.$message->reject(false)— Rabbit manda direto para a.dlq, sem reentrega. Use quando o erro é claramente não recuperável (payload inválido,message_typedesconhecido).$message->nack(true)— Rabbit reentrega a mensagem na mesma fila. Use para falhas transitórias (timeout, serviço externo fora do ar). A cada reentrega o contador de tentativas da fila quorum sobe; ao ultrapassarx-delivery-limit, o próprio Rabbit manda a mensagem para a.dlqautomaticamente — não é preciso contar tentativas manualmente no código do consumer.
Isso é o que evita o loop infinito de nack(true): sem x-delivery-limit, uma
mensagem "envenenada" fica sendo reentregue para sempre e trava o
processamento das mensagens atrás dela na fila; com o limite configurado, ela
sai do caminho principal após N tentativas e vai para a .dlq, onde pode ser
inspecionada/reprocessada manualmente sem impactar a fila principal.
Para desabilitar (não recomendado em produção) ou customizar por prioridade:
'dead_letter' => [ 'enabled' => true, 'suffix' => '.dlq', 'delivery_limit' => 3, 'queue_type' => 'quorum', // tipo da fila .dlq em si 'priorities' => [ 'high' => [], // usa os valores globais acima 'default' => ['delivery_limit' => 5], // sobrescreve só o delivery_limit 'low' => ['enabled' => false], // desliga DLQ só para low ], ],
Retry com atraso (TTL) é outro mecanismo, separado da DLQ. O padrão
.retryusado hoje no Eco UTM (ex.:UpdateFlightCommand, fila.retrycomx-message-ttlpara reentregar depois de um tempo) resolve um problema diferente — espaçar retentativas no tempo — e não faz parte desta entrega. A DLQ aqui só garante que mensagens esgotadas ou irrecuperáveis não travem a fila principal; um mecanismo de retry com TTL genérico pode ser adicionado depois como uma segunda etapa, sem alterar o que já existe.