redis队列插件

v1.1.4 版本
2022-04-24 版本更新时间
4588 安装
10 star

简介

基于Redis的消息队列,支持消息延迟处理。

安装

composer require webman/redis-queue

配置文件

redis配置文件自动生成在 config/plugin/webman/redis-queue/redis.php,内容类似如下:

<?php
return [
    'default' => [
        'host' => 'redis://127.0.0.1:6379',
        'options' => [
            'auth' => '',         // 密码,可选参数
            'db' => 0,            // 数据库
            'max_attempts'  => 5, // 消费失败后,重试次数
            'retry_seconds' => 5, // 重试间隔,单位秒
        ]
    ],
];

消费失败重试

如果消费失败(发生了异常),则消息会放入延迟队列,等待下次重试。重试次数通过参数 max_attempts 控制,重试间隔由
retry_secondsmax_attempts共同控制。比如max_attempts为5,retry_seconds为10,第1次重试间隔为1*10秒,第2次重试时间间隔为 2*10秒,第3次重试时间间隔为3*10秒,以此类推直到重试5次。如果超过了max_attempts设置测重试次数,则消息放入key为{redis-queue}-failed的失败队列。

投递消息

<?php
namespace app\controller;

use support\Request;
use Webman\RedisQueue\Client;

class Index
{
    public function queue(Request $request)
    {
        // 队列名
        $queue = 'send_mail';
        // 数据,可以直接传数组,无需序列化
        $data = ['to' => 'tom@gmail.com', 'content' => 'hello'];
        // 投递消息
        Client::send($queue, $data);
        // 投递延迟消息,消息会在60秒后处理
        Client::send($queue, $data, 60);

        return response('redis queue test');
    }

}

延迟队列消费时间可能会出现误差,例如消费速度小于生产速度导致队列积压,进而导致消费延迟,缓解办法是多开一些消费进程。

在其他项目投递消息

有时候你需要在其它项目中投递消息并且无法使用webman\redis-queue,则可以参考以下函数向队列投递消息。

function redis_queue_send($redis, $queue, $data, $delay = 0) {
    $queue_waiting = '{redis-queue}-waiting';
    $queue_delay = '{redis-queue}-delayed';
    $now = time();
    $package_str = json_encode([
        'id'       => rand(),
        'time'     => $now,
        'delay'    => 0,
        'attempts' => 0,
        'queue'    => $queue,
        'data'     => $data
    ]);
    if ($delay) {
        return $redis->zAdd($queue_delay, $now + $delay, $package_str);
    }
    return $redis->lPush($queue_waiting.$queue, $package_str);
}

其中,参数$redis为redis实例。例如redis扩展用法类似如下:

$redis = new Redis;
$redis->connect('127.0.0.1', 6379);
$queue = 'user-1';
$data= ['some', 'data'];
redis_queue_send($redis, $queue, $data);

消费

消费进程配置文件在 config/plugin/webman/redis-queue/process.php
消费者目录在 app/queue/redis/ 下。

新建 app/queue/redis/MyMailSend.php (类名任意,符合psr4规范即可)。

<?php

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;

class MyMailSend implements Consumer
{
    // 要消费的队列名
    public $queue = 'send_mail';

    // 连接名,对应 plugin/webman/redis-queue/redis.php 里的连接`
    public $connection = 'default';

    // 消费
    public function consume($data)
    {
        // 无需反序列化
        var_export($data); // 输出 ['to' => 'tom@gmail.com', 'content' => 'hello']
    }
}

注意
消费过程中没有抛出异常视为消费成功,否则消费失败,进入重试队列

提示
消费进程可以同时消费多种不同的队列,新增队列不需要修改process.php中的配置,新增队列消费者时只需要在app/queue/redis下新增对应的Consumer类即可,并用类属性$queue指定要消费的队列名