PHP code example of dy / mq

    
        
<?php
require_once('vendor/autoload.php');

/* Start to develop here. Best regards https://php-download.com/ */

    

dy / mq example snippets

use Dy\MessageQueue\Message\Message;
use Dy\MessageQueue\Processor\ConsumerProcessor;

class DemoMessageProcessor implements ConsumerProcessor
{
    public function handle(Message $message): bool
    {
        var_dump($message->getData());
        return true;
    }
}

class DemoDelayMessageProcessor implements ConsumerProcessor
{
    public function handle(Message $message): bool
    {
        var_dump($message->getData());
        return true;
    }
}

// config/message_queue.php
return [
    // 默认MQ驱动，与connections对应
    'driver'        =>  env('MQ_DRIVER', 'amqp'),

    'connections'   =>  [
        // AMQP驱动
        'amqp'      =>  [
            'host'      =>  env('MQ_AMQP_HOST', '127.0.0.1'),
            'port'      =>  env('MQ_AMQP_PORT', 5672),
            'user'      =>  env('MQ_AMQP_USER', 'guest'),
            'password'  =>  env('MQ_AMQP_PASSWORD', 'guest'),
            'vhost'     =>  env('MQ_AMQP_VHOST', '/'),
        ],
        // Redis驱动
        'redis'     =>  [
            'host'      =>  env('MQ_REDIS_HOST', '127.0.0.1'),
            'port'      =>  env('MQ_REDIS_PORT', 6379),
            'password'  =>  env('MQ_REDIS_PASSWORD', ''),
            'database'  =>  env('MQ_REDIS_DATABASE', 0),
            'max_len'   =>  env('MQ_REDIS_MAX_LEN', 1000),
        ]
    ],

    'prefix'        =>  '', // key前缀

    'processor'     =>  [ // 实现 Dy\MessageQueue\Processor\ConsumerProcessor 接口的队列消息处理器，用于对接业务逻辑
        'demo'      =>  DemoMessageProcessor::class,
        'demo.delay'=>  DemoDelayMessageProcessor::class,
    ],
    
    'retry'         =>  3,  // 消息失败重试次数

    'log'           =>  [   // 日志
        'level'     =>  'debug',
        'file'      =>  storage_path('logs/dy_message_queue.log')
    ],
];

use \Dy\MessageQueue\Facade\MQ;

// 发送普通队列消息，消息将被立即投递到消费者
MQ::queue('test.exchange', 'test.queue', 'test', 'hello world');

// 发送延时队列消息，消息会在10秒后投递到消费者
MQ::delayQueue('test.delay.exchange', 'test.delay.queue', 'test.delay', 'hello world', 10);

shell
$ php artisan vendor:publish # 选择 Provider: Dy\MessageQueue\ServiceProvider