hzted123/yii2-amqp

a consumer manage tool for amqp

1.0.4 2017-05-10 11:59 UTC

Requires

• php: >=5.3.0
• hzted123/yii2-cronjobs: *
• php-amqplib/php-amqplib: ~2.0
• yiisoft/yii2: *

MIT

• songming

• extension
• AMQP
• yii2

README

A tool to amqp consumer. It can declare exchange and queue with config. And Keep consumer alive with cronjobs.

This extension is based on this. Thanks webtoucher.

But with a few change:

• consumer setting by queue, not route or exchange
• all exchange and queue setting in config
• add listener-manage controller

Installation

The preferred way to install this extension is through composer.

Either run

php composer.phar require --prefer-dist hzted123/yii2-amqp "*"

or add

"hzted123/yii2-amqp": "*"

to the require section of your composer.json file.

Usage

Once the extension is installed, simply use it in your code by :

Add the following in your console config:

return [
    ...
    'components' => [
        ...
        'amqp' => [
            'class' => 'hzted123\amqp\components\Amqp',
            'host' => '******',
            'port' => 5672,
            'user' => '******',
            'password' => '******',
            'vhost' => '/',
            'exchange_configs' => [
                'exchange_name' => [
                    'options' => ['type' => 'topic', 'passive' => false, 'auto_delete' => false, 'durable' => true ],
                ],
                ... ...
            ],
            'queue_configs' => [
                'queue_name'    =>  [
                    'options' => ['passive' => false, 'auto_delete' => false, 'durable' => true, 'exclusive' => false],
                    'arguments' => ['x-max-length' => ['I', 1000000], 'x-max-length-bytes' => ['I', 300485760]],
                    'binds' => ['route' => 'exchange_name']
                ],
                ... ... 
            ],
        ],
    ],
    ...
    'controllerMap' => [
        'cron' => [
            'class' => 'mitalcoi\cronjobs\CronController',
            'interpreterPath' => '/usr/bin/php',
            'logsDir' => '/data/logs/cron',
            'logFileName' => '%L/php-%C.%A.%D.log',
            'bootstrapScript' => (dirname(dirname(__FILE__)) .'/yii',
            'cronJobs' =>[
                'listener-manage/keep' => [
                    'cron'      => '* * * * *',
                ]
            ],
        ],
        'listener' => [
            'class' => 'hzted123\amqp\controllers\AmqpListenerController',
            'interpreters' => [
                'queue_name' => '@app\components\DemoEventInterpreter', // interpreters for each queue
            ],
        ],
        'listener-manage' => [      //consumer keeper
            'class' => 'hzted123\amqp\controllers\ListenerManageController',
            'configs' => [
                ['queue' => 'queue_name', 'count' => 2]      // Keeping the number of consumers
            ]
        ],
    ],
];

Add messages interpreter class @app/components/DemoEventInterpreter with your handlers for different routing keys:

<?php

namespace app\components;

use hzted123\amqp\components\AmqpInterpreter;


class DemoEventInterpreter extends AmqpInterpreter
{
    /**
     * Interprets AMQP message with routing key 'hello_world'.
     *
     * @param array $message
     */
    public function readHelloWorld($message)
    {
        // todo: write message handler
        $this->log(print_r($message, true));
    }
}

Usage

Just run command

$ php yii listener-manage keep

to start all consumers , or like this

* * * * * php yii listener-manage keep

Run command

$ php yii listener-manage kill

to kill all consumers, when you deploy new code.

Run command

$ php yii listener --queue=queue_name

to listen a queue on selected queue.

Also you can create controllers for your needs. Just use for your web controllers class hzted123\amqp\controllers\AmqpConsoleController instead of yii\web\Controller and for your console controllers class hzted123\amqp\controllers\AmqpConsoleController instead of yii\console\Controller. AMQP connection will be available with property connection. AMQP channel will be available with property channel.

Note: The configuration information, starting with version 1.0.2, exchange and route switching position, for multiple routes binding to a queue

    'queue_configs' => [
        'queue_name'    =>  [
            'options' => ['passive' => false, 'auto_delete' => false, 'durable' => true, 'exclusive' => false],
            'arguments' => ['x-max-length' => ['I', 1000000], 'x-max-length-bytes' => ['I', 300485760]],
            'binds' => ['route' => 'exchange_name']
        ],
        ... ... 
    ],