README

Laravel 和 Lumen 的 AMQP 封装器，特别用于从 RabbitMQ 发布和消费消息

特性

• 高级队列配置
• 轻松将消息添加到队列
• 使用有用选项监听队列

安装

Composer

将以下内容添加到 composer.json 文件的 require 部分

"filld/larabbit": "2.*"

$ php composer update

或

$ php composer require filld/larabbit

集成

环境变量

以下环境变量需要在您的 .env 文件中设置

RABBITMQ_HOST
RABBITMQ_PORT
RABBITMQ_USERNAME
RABBITMQ_PASSWORD
RABBITMQ_VHOST

# Optional SSL:
RABBITMQ_SSL_CERT_PATH

Lumen

在您的 Lumen 应用程序的根目录下创建一个 config 文件夹，并将 vendor/filld/larabbit/config/amqp.php 的内容复制到 config/amqp.php。

根据您的需求调整属性。

return [

    'use' => 'production',

    'properties' => [

        'production' => [
            'host'                => env('RABBITMQ_HOST', 'localhost'),
            'port'                => env('RABBITMQ_PORT', 5672),
            'username'            => env('RABBITMQ_USERNAME'),
            'password'            => env('RABBITMQ_PASSWORD'),
            'vhost'               => env('RABBITMQ_VHOST'),
            'exchanges'           => [
                                         [
                                             'exchange'              => 'exchange_name',
                                             'exchange_type'         => 'topic',
                                             'exchange_passive'      => false,
                                             'exchange_durable'      => true,
                                             'exchange_auto_delete'  => false,
                                             'exchange_internal'     => false,
                                             'exchange_nowait'       => false,
                                             'exchange_properties'   => [],
                         
                                             'routing' => [
                                                 'routing.key.one',
                                                 'routing.key.two',
                                         ]
                                     ],
            'consumer_tag'        => 'consumer',
            'ssl_options'         => [], // See https://secure.php.net/manual/en/context.ssl.php
            'connect_options'     => [], // See https://github.com/php-amqplib/php-amqplib/blob/master/PhpAmqpLib/Connection/AMQPSSLConnection.php
            'queue_properties'    => ['x-ha-policy' => ['S', 'all']],
            'timeout'             => 0
        ],

    ],

];

在 bootstrap/app.php 中注册 Lumen 服务提供者

/*
|--------------------------------------------------------------------------
| Register Service Providers
|--------------------------------------------------------------------------
*/

//...

$app->configure('amqp');
$app->register(Filld\Amqp\LumenServiceProvider::class);

//...

启用 Lumen 5.2+ 的 Facade 支持

$app->withFacades();

Laravel

打开 config/app.php 并添加服务提供者和别名

'Filld\Amqp\AmqpServiceProvider',

'Amqp' => 'Filld\Amqp\Facades\Amqp',

设置 SSL

请确保将 CA 证书链复制到您可以引用的位置。建议使用 storage/certs 目录。同时，别忘了在配置中设置 SSL 设置

'connect_options'     => [
    'capath' => '/etc/ssl/certs',
    'cafile' => env('RABBITMQ_SSL_CERT', storage_path() . '/certs/cacert.pem'),
    'verify_peer' => true
],

同时，别忘了端口可能已更改为 5671

发布消息

使用路由键推送消息

    Amqp::publish('routing-key', 'message');

使用路由键推送消息并创建队列

    Amqp::publish('routing-key', 'message' , ['queue' => 'queue-name']);

使用路由键推送消息并覆盖属性

    Amqp::publish('routing-key', 'message' , ['exchange' => 'amq.direct']);

请注意，如果您尝试设置如 x-message-ttl 这样的属性，您可能会收到以下错误

AMQP-rabbit 没有定义 [] 类型的数据

您 必须 指定一个类型

    Amqp::publish('routing-key', 'message', [
        'queue_properties' => [
            "x-ha-policy" => ["S", "all"],
            'x-message-ttl' => ['I', 86400000],
            'x-dead-letter-exchange' => ['S', 'orders_dead_letter'],
        ],
        'queue' => 'orders'
    ]);

消费消息

消费消息，确认并停止，当没有消息时

Amqp::consume('queue-name', function ($message, $resolver) {
    		
   var_dump($message->body);

   $resolver->acknowledge($message);

   $resolver->stopWhenProcessed();
        
});

永久消费消息

Amqp::consume('queue-name', function ($message, $resolver) {
    		
   var_dump($message->body);

   $resolver->acknowledge($message);
        
});

使用自定义设置消费消息

Amqp::consume('queue-name', function ($message, $resolver) {
    		
   var_dump($message->body);

   $resolver->acknowledge($message);
      
}, [
	'timeout' => 2,
	'vhost'   => 'vhost3'
]);

Fanout 示例

发布消息

\Amqp::publish('', 'message' , [
    'exchange_type' => 'fanout',
    'exchange' => 'amq.fanout',
]);

消费消息

\Amqp::consume('', function ($message, $resolver) {
    var_dump($message->body);
    $resolver->acknowledge($message);
}, [
    'exchange' => 'amq.fanout',
    'exchange_type' => 'fanout',
    'queue_force_declare' => true,
    'queue_exclusive' => true,
    'persistent' => true// required if you want to listen forever
]);

致谢

• 该项目是 https://github.com/bschmitt/laravel-amqp 的分支
• 一些概念来自 https://github.com/mookofe/tail

许可证

本软件包是开源软件，根据 MIT 许可证 授权