redis队列插件

v1.3.2 版本
2024-04-03 版本更新时间
54973 安装
26 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的失败队列。

投递消息(同步)

注意
需要webman/redis >= 1.2.0,依赖 redis扩展

<?php
namespace app\controller;

use support\Request;
use Webman\RedisQueue\Redis;

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

        return response('redis queue test');
    }

}

投递成功Redis::send() 返回true,否则返回false或者抛出异常。

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

投递消息(异步)

<?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');
    }

}

Client::send() 没有返回值,它属于异步推送,它不保证消息%100送达redis。

提示
Client::send()原理是在本地内存建立一个内存队列,异步将消息同步到redis(同步速度很快,每秒大概1万笔消息)。如果进程重启,恰好本地内存队列里数据没有同步完毕,会造成消息丢失。Client::send()异步投递适合投递不重要的消息。

提示
Client::send()是异步的,它只能在workerman的运行环境中使用,命令行脚本请使用同步接口Redis::send()

在其他项目投递消息

有时候你需要在其它项目中投递消息并且无法使用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'    => $delay,
        '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/ 下。

执行命令php webman redis-queue:consumer my-send-mail则会生成文件app/queue/redis/MyMailSend.php

提示
如果命令不存在也可以手动生成

<?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']
    }

}

注意
消费过程中没有抛出异常和Error视为消费成功,否则消费失败,进入重试队列。
redis-queue没有ack机制,你可以把它看作是自动ack(没有产生异常或Error)。如果消费过程中想标记当前消息消费不成功,可以手动抛出异常,让当前消息进入重试队列。这实际上和ack机制没有区别。

提示
消费者支持多服务器多进程,并且同一条消息不会被重复消费。消费过的消息会自动从队列删除,无需手动删除。

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

提示
windows用户需要执行php windows.php 启动webman,否则不会启动消费进程

自定义失败处理

当开发者想介入消费失败后的处理流程时,可以实现onConsumeFailure方法,例如

<?php

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;

class MyMailSend implements Consumer
{
    public $queue = 'send-mail';
    public $connection = 'default';

    // 消费
    public function consume($data)
    {
        // 省略
    }

    // 消费失败时
    public function onConsumeFailure(\Throwable $exception, $package)
    {
        var_export($package);
        // 直接更改消息队列数据结构,将最大重试次数max_attempts字段设置为0,即不再重试。
        $package['max_attempts'] = 0;
        // 除此之外还可更改data字段(也就是consume方法中的$data)

        // 返回更改后的数据结构
        return $package;
    }

}

为某些队列单独设置消费进程

默认情况下,所有的消费者共用相同的消费进程。但有时我们需要将一些队列的消费独立出来,例如消费慢的业务放到一组进程中消费,消费快的业务放到另外一组进程消费。为此我们可以将消费者分为两个目录,例如 app_path() . '/queue/redis/fast'app_path() . '/queue/redis/slow' (注意消费类的命名空间需要做相应的更改),则配置如下:

return [
    ...这里省略了其它配置...

    'redis_consumer_fast'  => [
        'handler'     => Webman\RedisQueue\Process\Consumer::class,
        'count'       => 8,
        'constructor' => [
            // 消费者类目录
            'consumer_dir' => app_path() . '/queue/redis/fast'
        ]
    ],
    'redis_consumer_slow'  => [
        'handler'     => Webman\RedisQueue\Process\Consumer::class,
        'count'       => 8,
        'constructor' => [
            // 消费者类目录
            'consumer_dir' => app_path() . '/queue/redis/slow'
        ]
    ]
];

这样快业务消费者放到queue/redis/fast目录下,慢业务消费者放到queue/redis/slow目录下达到给队列指定消费进程的目的。

多redis配置

配置

config/plugin/webman/redis-queue/redis.php

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

注意配置里增加了一个other为key的redis配置

多redis投递消息

// 向 `default` 为key的队列投递消息
Client::connection('default')->send($queue, $data);
Redis::connection('default')->send($queue, $data);
//  等同于
Client::send($queue, $data);
Redis::send($queue, $data);

// 向 `other` 为key的队列投递消息
Client::connection('other')->send($queue, $data);
Redis::connection('other')->send($queue, $data);

多redis消费

消费配置里other 为key的队列投递消息

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;

class SendMail implements Consumer
{
    // 要消费的队列名
    public $queue = 'send-mail';

    // === 这里设置为other,代表消费配置里other为key的队列 ===
    public $connection = 'other';

    // 消费
    public function consume($data)
    {
        // 无需反序列化
        var_export($data);
    }
}

常见问题

为什么会有报错 Workerman\Redis\Exception: Workerman Redis Wait Timeout (600 seconds)

这个错误只会存在于异步投递接口Client::send()中。异步投递首先会将消息保存在本地内存中,当进程空闲时将消息发送给redis。如果redis接收速度慢于消息生产速度,或者进程一直忙于其它业务没有足够的时间将内存的消息同步给redis,就会导致消息挤压。如果有消息挤压超过600秒,就会触发此错误。

解决方案:投递消息使用同步投递接口Redis::send()

赞助商