laravel queue rabbitmq
v14.1.0
Apenas a versão mais recente terá novos recursos. Correções de bugs serão fornecidas usando o seguinte esquema:
| Versão do pacote | Versão Laravel | Correções de bugs até | |
|---|---|---|---|
| 13 | 9 | 8 de agosto de 2023 | Documentação |
Você pode instalar este pacote via compositor usando este comando:
composer require vladimir-yuldashev/laravel-queue-rabbitmq
O pacote será registrado automaticamente.
Adicione conexão a config/queue.php :
Esta é a configuração mínima para que a conexão/driver RabbitMQ funcione.
' connections ' => [
// ...
' rabbitmq ' => [
' driver ' => ' rabbitmq ' ,
' hosts ' => [
[
' host ' => env ( ' RABBITMQ_HOST ' , ' 127.0.0.1 ' ),
' port ' => env ( ' RABBITMQ_PORT ' , 5672 ),
' user ' => env ( ' RABBITMQ_USER ' , ' guest ' ),
' password ' => env ( ' RABBITMQ_PASSWORD ' , ' guest ' ),
' vhost ' => env ( ' RABBITMQ_VHOST ' , ' / ' ),
],
// ...
],
// ...
],
// ...
],Opcionalmente, adicione opções de fila à configuração de uma conexão. Cada fila criada para esta conexão obtém as propriedades.
Quando você deseja priorizar mensagens quando elas estão atrasadas, isso é possível adicionando opções extras.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' options ' => [
' queue ' => [
// ...
' prioritize_delayed ' => false ,
' queue_max_priority ' => 10 ,
],
],
],
// ...
],Quando você deseja publicar mensagens em uma troca com chaves de roteamento, isso é possível adicionando opções extras.
amq.direct para a chave de roteamentoqueue .%s na chave de roteamento, o queue_name será substituído.Nota: ao usar uma troca com chave de roteamento, você provavelmente cria suas filas com ligações.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' options ' => [
' queue ' => [
// ...
' exchange ' => ' application-x ' ,
' exchange_type ' => ' topic ' ,
' exchange_routing_key ' => '' ,
],
],
],
// ...
],No Laravel, os trabalhos com falha são armazenados no banco de dados. Mas talvez você queira instruir algum outro processo para também fazer algo com a mensagem. Quando você deseja instruir o RabbitMQ a redirecionar mensagens com falha para uma troca ou fila específica, isso é possível adicionando opções extras.
amq.direct para a chave de roteamentoqueue é substituído por '.failed' .%s na chave de roteamento, o queue_name será substituído.Nota: Ao usar a troca failed_job com a chave de roteamento, você provavelmente precisará criar sua troca/fila com ligações você mesmo.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' options ' => [
' queue ' => [
// ...
' reroute_failed ' => true ,
' failed_exchange ' => ' failed-exchange ' ,
' failed_routing_key ' => ' application-x.%s ' ,
],
],
],
// ...
], A partir da versão 8.0, este pacote oferece suporte imediato ao Laravel Horizon. Primeiramente, instale o Horizon e defina RABBITMQ_WORKER como horizon .
Horizon depende de eventos despachados pelo trabalhador. Esses eventos informam ao Horizon o que foi feito com a mensagem/trabalho.
Esta Biblioteca suporta Horizon, mas na configuração você tem que informar ao Laravel para usar o QueueApi compatível com Horizon.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
/* Set to "horizon" if you wish to use Laravel Horizon. */
' worker ' => env ( ' RABBITMQ_WORKER ' , ' default ' ),
],
// ...
], Às vezes você precisa trabalhar com mensagens publicadas por outro aplicativo.
Essas mensagens provavelmente não respeitarão o esquema de carga útil do trabalho do Laravel. O problema com essas mensagens é que os trabalhadores do Laravel não serão capazes de determinar o trabalho ou classe real a ser executado.
Você pode estender o RabbitMQJob::class integrado e dentro da configuração de conexão da fila, você pode definir sua própria classe. Quando você especifica uma chave job na configuração, com seu próprio nome de classe, cada mensagem recuperada do broker será empacotada por sua própria classe.
Um exemplo para a configuração:
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' options ' => [
' queue ' => [
// ...
' job ' => App Queue Jobs RabbitMQJob::class,
],
],
],
// ...
],Um exemplo de sua própria classe de trabalho:
<?php
namespace App Queue Jobs ;
use VladimirYuldashev LaravelQueueRabbitMQ Queue Jobs RabbitMQJob as BaseJob ;
class RabbitMQJob extends BaseJob
{
/**
* Fire the job.
*
* @return void
*/
public function fire ()
{
$ payload = $ this -> payload ();
$ class = WhatheverClassNameToExecute::class;
$ method = ' handle ' ;
( $ this -> instance = $ this -> resolve ( $ class ))->{ $ method }( $ this , $ payload );
$ this -> delete ();
}
}Ou talvez você queira adicionar propriedades extras à carga:
<?php
namespace App Queue Jobs ;
use VladimirYuldashev LaravelQueueRabbitMQ Queue Jobs RabbitMQJob as BaseJob ;
class RabbitMQJob extends BaseJob
{
/**
* Get the decoded body of the job.
*
* @return array
*/
public function payload ()
{
return [
' job ' => ' WhatheverFullyQualifiedClassNameToExecute@handle ' ,
' data ' => json_decode ( $ this -> getRawBody (), true )
];
}
} Se você deseja lidar com mensagens brutas, não no formato JSON ou sem a chave 'job' em JSON, você deve adicionar stub para o método getName :
<?php
namespace App Queue Jobs ;
use Illuminate Support Facades Log ;
use VladimirYuldashev LaravelQueueRabbitMQ Queue Jobs RabbitMQJob as BaseJob ;
class RabbitMQJob extends BaseJob
{
public function fire ()
{
$ anyMessage = $ this -> getRawBody ();
Log:: info ( $ anyMessage );
$ this -> delete ();
}
public function getName ()
{
return '' ;
}
} Você pode estender o PhpAmqpLibConnectionAMQPStreamConnection::class ou PhpAmqpLibConnectionAMQPSLLConnection::class e dentro da configuração da conexão, você pode definir sua própria classe. Quando você especifica uma chave connection na configuração, com seu próprio nome de classe, cada conexão usará sua própria classe.
Um exemplo para a configuração:
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' connection ' = > App Queue Connection MyRabbitMQConnection::class,
],
// ...
], Se você quiser usar seu próprio RabbitMQQueue::class isso é possível estendendo VladimirYuldashevLaravelQueueRabbitMQQueueRabbitMQQueue . e informe o laravel para usar sua classe definindo RABBITMQ_WORKER como AppQueueRabbitMQQueue::class .
Nota: As classes de trabalho devem estender
VladimirYuldashevLaravelQueueRabbitMQQueueRabbitMQQueue
' connections ' => [
// ...
' rabbitmq ' => [
// ...
/* Set to a class if you wish to use your own. */
' worker ' => App Queue RabbitMQQueue::class,
],
// ...
], <?php
namespace App Queue ;
use VladimirYuldashev LaravelQueueRabbitMQ Queue RabbitMQQueue as BaseRabbitMQQueue ;
class RabbitMQQueue extends BaseRabbitMQQueue
{
// ...
}Por exemplo: uma implementação de reconexão.
Se você deseja se reconectar ao RabbitMQ, se a conexão estiver inoperante. Você pode substituir os métodos de publicação e createChannel.
Nota: esta não é uma prática recomendada, é um exemplo.
<?php
namespace App Queue ;
use PhpAmqpLib Exception AMQPChannelClosedException ;
use PhpAmqpLib Exception AMQPConnectionClosedException ;
use VladimirYuldashev LaravelQueueRabbitMQ Queue RabbitMQQueue as BaseRabbitMQQueue ;
class RabbitMQQueue extends BaseRabbitMQQueue
{
protected function publishBasic ( $ msg , $ exchange = '' , $ destination = '' , $ mandatory = false , $ immediate = false , $ ticket = null ): void
{
try {
parent :: publishBasic ( $ msg , $ exchange , $ destination , $ mandatory , $ immediate , $ ticket );
} catch ( AMQPConnectionClosedException | AMQPChannelClosedException ) {
$ this -> reconnect ();
parent :: publishBasic ( $ msg , $ exchange , $ destination , $ mandatory , $ immediate , $ ticket );
}
}
protected function publishBatch ( $ jobs , $ data = '' , $ queue = null ): void
{
try {
parent :: publishBatch ( $ jobs , $ data , $ queue );
} catch ( AMQPConnectionClosedException | AMQPChannelClosedException ) {
$ this -> reconnect ();
parent :: publishBatch ( $ jobs , $ data , $ queue );
}
}
protected function createChannel (): AMQPChannel
{
try {
return parent :: createChannel ();
} catch ( AMQPConnectionClosedException ) {
$ this -> reconnect ();
return parent :: createChannel ();
}
}
}A conexão usa uma fila padrão com valor 'default', quando nenhuma fila é fornecida pelo laravel. É possível alterar a fila padrão adicionando um parâmetro extra na configuração da conexão.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' queue ' => env ( ' RABBITMQ_QUEUE ' , ' default ' ),
],
// ...
], Por padrão, sua conexão será criada com uma configuração de pulsação igual a 0 . Você pode alterar as configurações de pulsação alterando o arquivo config.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' options ' => [
// ...
' heartbeat ' => 10 ,
],
],
// ...
],Se precisar de uma conexão segura com o(s) servidor(es) RabbitMQ, você precisará adicionar essas opções de configuração extras.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' secure ' = > true ,
' options ' => [
// ...
' ssl_options ' => [
' cafile ' => env ( ' RABBITMQ_SSL_CAFILE ' , null ),
' local_cert ' => env ( ' RABBITMQ_SSL_LOCALCERT ' , null ),
' local_key ' => env ( ' RABBITMQ_SSL_LOCALKEY ' , null ),
' verify_peer ' => env ( ' RABBITMQ_SSL_VERIFY_PEER ' , true ),
' passphrase ' => env ( ' RABBITMQ_SSL_PASSPHRASE ' , null ),
],
],
],
// ...
],Para instruir os trabalhadores do Laravel a despachar eventos após a conclusão de todos os commits do banco de dados.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' after_commit ' => true ,
],
// ...
],Por padrão, sua conexão será criada como uma conexão lenta. Se por algum motivo você não quiser que a conexão seja lenta, você pode desligá-la definindo a seguinte configuração.
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' lazy ' = > false ,
],
// ...
], Por padrão, o protocolo de rede usado para conexão é o tcp. Se por algum motivo você quiser usar outro protocolo de rede, poderá adicionar o valor extra em suas opções de configuração. Protocolos disponíveis: tcp , ssl , tls
' connections ' => [
// ...
' rabbitmq ' => [
// ...
' network_protocol ' => ' tcp ' ,
],
// ...
],A partir da versão 13.3.0, este pacote oferece suporte ao Laravel Octane pronto para uso. Em primeiro lugar, instale o Octane e não se esqueça de aquecer a conexão 'rabbitmq' na configuração do octane.
Veja: # 460 (comentário)
Depois de concluir a configuração, você pode usar a API Laravel Queue. Se você usou outros drivers de fila, não será necessário alterar mais nada. Se você não sabe como usar a API Queue, consulte a documentação oficial do Laravel: http://laravel.com/docs/queues
Para uso do Lumen, o provedor de serviços deve ser registrado manualmente conforme segue em bootstrap/app.php :
$ app -> register ( VladimirYuldashev LaravelQueueRabbitMQ LaravelQueueRabbitMQServiceProvider::class);Existem duas maneiras de consumir mensagens.
comando queue:work que é o comando interno do Laravel. Este comando utiliza basic_get . Use isto se quiser consumir várias filas.
comando rabbitmq:consume que é fornecido por este pacote. Este comando utiliza basic_consume e tem melhor desempenho que basic_get em aproximadamente 2x, mas não oferece suporte a múltiplas filas.
Configure o RabbitMQ usando docker-compose :
docker compose up -dPara executar o conjunto de testes você pode usar os seguintes comandos:
# To run both style and unit tests.
composer test
# To run only style tests.
composer test:style
# To run only unit tests.
composer test:unitSe você receber algum erro nos testes de estilo, poderá corrigir automaticamente a maioria, senão todos os problemas, com o seguinte comando:
composer fix:styleVocê pode contribuir com este pacote descobrindo bugs e abrindo problemas. Por favor, adicione em qual versão do pacote você cria pull request ou problema. (por exemplo, [5.2] Erro fatal em trabalho atrasado)
