1、安装

composer require workerman/redis-queue

2、使用

新增队列配置文件 Applications/config/queue.php

<?php
return [
    'type'          => 'redis',   //类型[固定]
    'host'          => '127.0.0.1',   //地址
    'port'          => '6379',   //端口
    'password'      => '',   //密码
    'select'        => 2,   //数据库db
    'max_attempts'  => 5,   //消费失败后重试次数
    'retry_seconds' => 5,   //重试时间间隔
];

新增队列消费者文件 Applications/queue/start_queue.php

<?php
use Workerman\Worker;
use Workerman\RedisQueue\Client;
$queue_config = require __DIR__ . '/../config/queue.php';

$worker = new Worker();
$worker->name = 'StartQueueWorker';
$worker->count = 2;

$worker->onWorkerStart = function () use ($queue_config) {
    $client = new Client($queue_config['type'].'://'.$queue_config['host'].':'.$queue_config['port'], [
        'auth'          => $queue_config['password'],       //密码
        'db'            => $queue_config['select'],         //数据库db
        'max_attempts'  => $queue_config['max_attempts'],   //消费失败后重试次数
        'retry_seconds' => $queue_config['retry_seconds'],  //重试时间间隔
    ]);
    
    $client->subscribe('user-1', function($data){
        echo "user-1\n";
        var_export($data);
    });
};

队列发送消息【在Events.php的onMessage里加入测试代码】

/*队列测试代码*/
$queue_config = require __DIR__ . '/../config/queue.php';
$client = new Client($queue_config['type'].'://'.$queue_config['host'].':'.$queue_config['port'], [
    'auth'          => $queue_config['password'],       //密码
    'db'            => $queue_config['select'],         //数据库db
    'max_attempts'  => $queue_config['max_attempts'],   //消费失败后重试次数
    'retry_seconds' => $queue_config['retry_seconds'],  //重试时间间隔
]);
$client->send('user-1', ['some', 'data']);
/*队列测试代码*/