Redis隊(duì)列

基于Redis的消息隊(duì)列，支持消息延遲處理。

安裝

composer require webman/redis-queue

配置文件

redis配置文件自動(dòng)生成在 {主項(xiàng)目}/config/plugin/webman/redis-queue/redis.php，內(nèi)容類似如下：

<?php
return [
    'default' => [
        'host' => 'redis://127.0.0.1:6379',
        'options' => [
            'auth' => '',         // 密碼，可選參數(shù)
            'db' => 0,            // 數(shù)據(jù)庫
            'max_attempts'  => 5, // 消費(fèi)失敗后，重試次數(shù)
            'retry_seconds' => 5, // 重試間隔，單位秒
        ]
    ],
];

消費(fèi)失敗重試

如果消費(fèi)失敗(發(fā)生了異常)，則消息會(huì)放入延遲隊(duì)列，等待下次重試。重試次數(shù)通過參數(shù) max_attempts 控制，重試間隔由
retry_seconds 和 max_attempts共同控制。比如max_attempts為5，retry_seconds為10，第1次重試間隔為1*10秒，第2次重試時(shí)間間隔為 2*10秒，第3次重試時(shí)間間隔為3*10秒，以此類推直到重試5次。如果超過了max_attempts設(shè)置測重試次數(shù)，則消息放入key為{redis-queue}-failed的失敗隊(duì)列。

投遞消息(同步)

<?php
namespace app\controller;

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

class Index
{
    public function queue(Request $request)
    {
        // 隊(duì)列名
        $queue = 'send-mail';
        // 數(shù)據(jù)，可以直接傳數(shù)組，無需序列化
        $data = ['to' => 'tom@gmail.com', 'content' => 'hello'];
        // 投遞消息
        Redis::send($queue, $data);
        // 投遞延遲消息，消息會(huì)在60秒后處理
        Redis::send($queue, $data, 60);

        return response('redis queue test');
    }

}

投遞成功Redis::send() 返回true，否則返回false或者拋出異常。

提示
延遲隊(duì)列消費(fèi)時(shí)間可能會(huì)出現(xiàn)誤差，例如消費(fèi)速度小于生產(chǎn)速度導(dǎo)致隊(duì)列積壓，進(jìn)而導(dǎo)致消費(fèi)延遲，緩解辦法是多開一些消費(fèi)進(jìn)程。

投遞消息(異步)

<?php
namespace app\controller;

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

class Index
{
    public function queue(Request $request)
    {
        // 隊(duì)列名
        $queue = 'send-mail';
        // 數(shù)據(jù)，可以直接傳數(shù)組，無需序列化
        $data = ['to' => 'tom@gmail.com', 'content' => 'hello'];
        // 投遞消息
        Client::send($queue, $data);
        // 投遞延遲消息，消息會(huì)在60秒后處理
        Client::send($queue, $data, 60);

        return response('redis queue test');
    }

}

Client::send() 沒有返回值，它屬于異步推送，它不保證消息%100送達(dá)redis。

提示
Client::send()原理是在本地內(nèi)存建立一個(gè)內(nèi)存隊(duì)列，異步將消息同步到redis(同步速度很快，每秒大概1萬筆消息)。如果進(jìn)程重啟，恰好本地內(nèi)存隊(duì)列里數(shù)據(jù)沒有同步完畢，會(huì)造成消息丟失。Client::send()異步投遞適合投遞不重要的消息。

提示
Client::send()是異步的，它只能在workerman的運(yùn)行環(huán)境中使用，命令行腳本請(qǐng)使用同步接口Redis::send()

在其他項(xiàng)目投遞消息

有時(shí)候你需要在其它項(xiàng)目中投遞消息并且無法使用webman\redis-queue，則可以參考以下函數(shù)向隊(duì)列投遞消息。

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

其中，參數(shù)$redis為redis實(shí)例。例如redis擴(kuò)展用法類似如下：

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

消費(fèi)

消費(fèi)進(jìn)程配置文件在 {主項(xiàng)目}/config/plugin/webman/redis-queue/process.php。
消費(fèi)者目錄在 {主項(xiàng)目}/app/queue/redis/ 下。

執(zhí)行命令php webman redis-queue:consumer my-send-mail則會(huì)生成文件{主項(xiàng)目}/app/queue/redis/MyMailSend.php

提示
如果命令不存在也可以手動(dòng)生成

<?php

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;

class MyMailSend implements Consumer
{
    // 要消費(fèi)的隊(duì)列名
    public $queue = 'send-mail';

    // 連接名，對(duì)應(yīng) plugin/webman/redis-queue/redis.php 里的連接`
    public $connection = 'default';

    // 消費(fèi)
    public function consume($data)
    {
        // 無需反序列化
        var_export($data); // 輸出 ['to' => 'tom@gmail.com', 'content' => 'hello']
    }
    // 消費(fèi)失敗回調(diào)
    /* 
    $package = [
        'id' => 1357277951, // 消息ID
        'time' => 1709170510, // 消息時(shí)間
        'delay' => 0, // 延遲時(shí)間
        'attempts' => 2, // 消費(fèi)次數(shù)
        'queue' => 'send-mail', // 隊(duì)列名
        'data' => ['to' => 'tom@gmail.com', 'content' => 'hello'], // 消息內(nèi)容
        'max_attempts' => 5, // 最大重試次數(shù)
        'error' => '錯(cuò)誤信息' // 錯(cuò)誤信息
    ]
    */
    public function onConsumeFailure(\Throwable $e, $package)
    {
        echo "consume failure\n";
        echo $e->getMessage() . "\n";
        // 無需反序列化
        var_export($package); 
    }
}

注意
消費(fèi)過程中沒有拋出異常和Error視為消費(fèi)成功，否則消費(fèi)失敗，進(jìn)入重試隊(duì)列。
redis-queue沒有ack機(jī)制，你可以把它看作是自動(dòng)ack(沒有產(chǎn)生異常或Error)。如果消費(fèi)過程中想標(biāo)記當(dāng)前消息消費(fèi)不成功，可以手動(dòng)拋出異常，讓當(dāng)前消息進(jìn)入重試隊(duì)列。這實(shí)際上和ack機(jī)制沒有區(qū)別。

提示
消費(fèi)者支持多服務(wù)器多進(jìn)程，并且同一條消息不會(huì)被重復(fù)消費(fèi)。消費(fèi)過的消息會(huì)自動(dòng)從隊(duì)列刪除，無需手動(dòng)刪除。

提示
消費(fèi)進(jìn)程可以同時(shí)消費(fèi)多種不同的隊(duì)列，新增隊(duì)列不需要修改process.php中的配置，新增隊(duì)列消費(fèi)者時(shí)只需要在app/queue/redis下新增對(duì)應(yīng)的Consumer類即可，并用類屬性$queue指定要消費(fèi)的隊(duì)列名

提示
windows用戶需要執(zhí)行php windows.php 啟動(dòng)webman，否則不會(huì)啟動(dòng)消費(fèi)進(jìn)程

提示
onConsumeFailure回調(diào)會(huì)在每次消費(fèi)失敗時(shí)觸發(fā)，你可以在這里處理失敗后的邏輯。

為不同的隊(duì)列設(shè)置不同的消費(fèi)進(jìn)程

默認(rèn)情況下，所有的消費(fèi)者共用相同的消費(fèi)進(jìn)程。但有時(shí)我們需要將一些隊(duì)列的消費(fèi)獨(dú)立出來，例如消費(fèi)慢的業(yè)務(wù)放到一組進(jìn)程中消費(fèi)，消費(fèi)快的業(yè)務(wù)放到另外一組進(jìn)程消費(fèi)。為此我們可以將消費(fèi)者分為兩個(gè)目錄，例如 app_path() . '/queue/redis/fast' 和 app_path() . '/queue/redis/slow' （注意消費(fèi)類的命名空間需要做相應(yīng)的更改），則配置如下：

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

    'redis_consumer_fast'  => [ // key是自定義的，沒有格式限制，這里取名redis_consumer_fast
        'handler'     => Webman\RedisQueue\Process\Consumer::class,
        'count'       => 8,
        'constructor' => [
            // 消費(fèi)者類目錄
            'consumer_dir' => app_path() . '/queue/redis/fast'
        ]
    ],
    'redis_consumer_slow'  => [  // key是自定義的，沒有格式限制，這里取名redis_consumer_slow
        'handler'     => Webman\RedisQueue\Process\Consumer::class,
        'count'       => 8,
        'constructor' => [
            // 消費(fèi)者類目錄
            'consumer_dir' => app_path() . '/queue/redis/slow'
        ]
    ]
];

這樣快業(yè)務(wù)消費(fèi)者放到queue/redis/fast目錄下，慢業(yè)務(wù)消費(fèi)者放到queue/redis/slow目錄下達(dá)到給隊(duì)列指定消費(fèi)進(jìn)程的目的。

多redis配置

配置

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

<?php
return [
    'default' => [
        'host' => 'redis://192.168.0.1:6379',
        'options' => [
            'auth' => null,       // 密碼，字符串類型，可選參數(shù)
            'db' => 0,            // 數(shù)據(jù)庫
            'max_attempts'  => 5, // 消費(fèi)失敗后，重試次數(shù)
            'retry_seconds' => 5, // 重試間隔，單位秒
        ]
    ],
    'other' => [
        'host' => 'redis://192.168.0.2:6379',
        'options' => [
            'auth' => null,       // 密碼，字符串類型，可選參數(shù)
            'db' => 0,             // 數(shù)據(jù)庫
            'max_attempts'  => 5, // 消費(fèi)失敗后，重試次數(shù)
            'retry_seconds' => 5, // 重試間隔，單位秒
        ]
    ],
];

注意配置里增加了一個(gè)other為key的redis配置

多redis投遞消息

// 向 `default` 為key的隊(duì)列投遞消息
Client::connection('default')->send($queue, $data);
Redis::connection('default')->send($queue, $data);
//  等同于
Client::send($queue, $data);
Redis::send($queue, $data);

// 向 `other` 為key的隊(duì)列投遞消息
Client::connection('other')->send($queue, $data);
Redis::connection('other')->send($queue, $data);

多redis消費(fèi)

消費(fèi)配置里other 為key的隊(duì)列投遞消息

namespace app\queue\redis;

use Webman\RedisQueue\Consumer;

class SendMail implements Consumer
{
    // 要消費(fèi)的隊(duì)列名
    public $queue = 'send-mail';

    // === 這里設(shè)置為other，代表消費(fèi)配置里other為key的隊(duì)列 ===
    public $connection = 'other';

    // 消費(fèi)
    public function consume($data)
    {
        // 無需反序列化
        var_export($data);
    }
}

常見問題

為什么會(huì)有報(bào)錯(cuò) Workerman\Redis\Exception: Workerman Redis Wait Timeout (600 seconds)

這個(gè)錯(cuò)誤只會(huì)存在于異步投遞接口Client::send()中。異步投遞首先會(huì)將消息保存在本地內(nèi)存中，當(dāng)進(jìn)程空閑時(shí)將消息發(fā)送給redis。如果redis接收速度慢于消息生產(chǎn)速度，或者進(jìn)程一直忙于其它業(yè)務(wù)沒有足夠的時(shí)間將內(nèi)存的消息同步給redis，就會(huì)導(dǎo)致消息擠壓。如果有消息擠壓超過600秒，就會(huì)觸發(fā)此錯(cuò)誤。

解決方案：投遞消息使用同步投遞接口Redis::send()。