README

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

功能

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

安装

Composer

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

"longhao/yii2-amqp": "1.*"

$ php composer update

或者

$ php composer require longhao/yii2-amqp

集成

Yii2

在您的 yii2 应用程序根目录中创建一个 config 文件夹，并将 vendor/longhao/yii2-amqp/config/amqp.php 的内容复制到 common/config/params-local.php。

根据需要调整属性。

return [
    'amqp' => [
        'use' => 'production',

        'properties' => [

            'production' => [
                'host'                => 'localhost',
                'port'                => 5672,
                'username'            => 'username',
                'password'            => 'password',
                'vhost'               => '/',
                'exchange'            => 'amq.topic',
                'exchange_type'       => 'topic',
                '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']],
                'exchange_properties' => [],
                'timeout'             => 0
            ],

        ],
    ]

];

在 common/config/main.php 中注册组件

/*
|--------------------------------------------------------------------------
| Register Component
|--------------------------------------------------------------------------
*/

//...
'components' => [
    'Amqp' => [
        'class' => 'Longhao\Amqp\Amqp'
    ]
]

//...

注册组件不是必需的。

Yii2

发布消息

使用路由键推送消息

示例 1

    $amqp = new Amqp;
    $amqp->publish('routing-key', 'message');

示例 2

    Yii::$app->Amqp->publish('routing-key', 'message');

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

示例 1

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

示例 2

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

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

示例 1

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

示例 2

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

消费消息

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

示例 1

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

   $resolver->acknowledge($message);

   $resolver->stopWhenProcessed();
        
});

示例 2

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

   $resolver->acknowledge($message);

   $resolver->stopWhenProcessed();
        
});

永久消费消息

示例 1

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

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

示例 2

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

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

使用自定义设置消费消息

示例 1

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

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

示例 2

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

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

Fanout 示例

发布消息

示例 1

$amqp = new Amqp;
$amqp->publish('', 'message' , [
    'exchange_type' => 'fanout',
    'exchange' => 'amq.fanout',
]);

示例 2

Yii::$app->Amqp->publish('', 'message' , [
    'exchange_type' => 'fanout',
    'exchange' => 'amq.fanout',
]);

消费消息

示例 1

$amqp = new Amqp;
$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
]);

示例 2

Yii::$app->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/mookofe/tail

许可协议

本软件包是开源软件，采用 MIT 许可协议 许可。