Redis 如何實作延時任務佇列

簡介

顧名思義，延遲佇列就是進入該佇列的訊息會被延遲消費的佇列。而一般的佇列，訊息一旦入隊了之後就會被消費者馬上消費。

延時任務和定時任務區別

業務場景

使用場景

延遲佇列多用於需要延遲工作的場景。

最常見的是以下兩種場景：

1、延遲消費

1. 使用者生成訂單之後，需要過一段時間校驗訂單的支付狀態，如果訂單仍未支付則需要及時地關閉訂單。

2. 使用者註冊成功之後，需要過一段時間比如一周後校驗使用者的使用情況，如果發現使用者活躍度較低，則發送信件或者簡訊來提醒使用者使用。

2、延遲重試

比如消費者從佇列裏消費訊息時失敗了，但是想要延遲一段時間後自動重試。如果不使用延遲佇列，那麽我們只能透過一個輪詢掃描程式去完成。

掃表存在的問題是

緩存佇列設計

場景設計

實際的生產場景是筆者負責的某個系統需要對接一個外部的資金方，每一筆資金下單後需要延時30分鐘推播對應的附件。

這裏簡化為一個訂單資訊數據延遲處理的場景，就是每一筆下單記錄一條訂單訊息(暫時叫做 OrderMessage )，訂單訊息需要延遲5到15秒後進行異步處理。

延時佇列的實作

選用了基於 Redis 的有序集合 Sorted Set 和 Crontab 短輪詢進行實作。

具體方案是：

1. 訂單建立的時候，訂單ID和當前時間戳分別作為 Sorted Set 的 member 和 score 添加到訂單佇列 Sorted Set 中。

2. 訂單建立的時候，訂單ID和推播內容 JSON 字串分別作為 field 和 value 添加到訂單佇列內容 Hash 中。

3. 第1步和第2步操作的時候用 Lua 指令碼保證原子性。

4. 使用一個異步執行緒透過 Sorted Set 的命令 ZREVRANGEBYSCORE 彈出指定數量的 訂單ID 對應的訂單佇列內容 Hash 中的訂單推播內容數據進行處理。

對於第4點處理有兩種方案：

處理方案一

彈出訂單內容數據的同時進行數據刪除，也就是 ZREVRANGEBYSCORE 、 ZREM 和 HDEL 命令要在同一個 Lua 指令碼中執行，這樣的話 Lua 指令碼的編寫難度大，並且由於彈出數據已經在 Redis 中刪除，如果數據處理失敗則可能需要從資料庫重新查詢補償。

處理方案二

彈出訂單內容數據之後，在數據處理完成的時候再主動刪除訂單佇列 Sorted Set 和訂單佇列內容 Hash 中對應的數據，這樣的話需要控制並行，有重復執行的可能性。

選用了方案一，也就是從 Sorted Set 彈出訂單ID並且從Hash中獲取完推播數據之後馬上刪除這兩個集合中對應的數據。

方案的流程圖大概是這樣：

Redis相關命令

Sorted Set相關命令

ZADD 命令 - 將一個或多個成員元素及其分數值加入到有序集當中。

ZADD KEY SCORE1 VALUE1.. SCOREN VALUEN


ZREVRANGEBYSCORE 命令 - 返回有序集中指定分數區間內的所有的成員。有序整合員按分數值遞減(從大到小)的次序排列。

ZREVRANGEBYSCORE key max min [WITHSCORES] [LIMIT offset count]


[success] ZREM 命令 - 用於移除有序集中的一個或多個成員，不存在的成員將被忽略。

ZREM key member [member ...]


Hash相關命令

HMSET 命令 - 同時將多個field-value(欄位-值)對設定到哈希表中。

HMSET KEY_NAME FIELD1 VALUE1 ...FIELDN VALUEN


HDEL 命令 - 刪除哈希表key中的一個或多個指定欄位，不存在的欄位將被忽略。

HDEL KEY_NAME FIELD1.. FIELDN


Lua 語法

如果不熟悉Lua語言，建議系統學習一下，因為想用好Redis，一定離不開Lua。

Lua 指令碼

入隊 enqueue.lua

local zset_key = KEYS[1]
local hash_key = KEYS[2]
local zset_value = ARGV[1]
local zset_score = ARGV[2]
local hash_field = ARGV[3]
local hash_value = ARGV[4]
redis.call('ZADD', zset_key, zset_score, zset_value)
redis.call('HSET', hash_key, hash_field, hash_value)
returnnil


將任務的執行時間作為score，要執行的任務數據作為value，存放在zset中

出隊 dequeue.lua

local zset_key = KEYS[1]
local hash_key = KEYS[2]
local min_score = ARGV[1]
local max_score = ARGV[2]
local offset = ARGV[3]
local limit = ARGV[4]
-- TYPE命令的返回結果是{'ok':'zset'}這樣子,這裏利用next做一輪叠代
localstatus, type = next(redis.call('TYPE', zset_key))
ifstatus ~= nilandstatus == 'ok'then
iftype == 'zset'then
local list = redis.call('ZREVRANGEBYSCORE', zset_key, max_score, min_score, 'LIMIT', offset, limit)
if list ~= niland #list > 0then
-- unpack函式能把table轉化為可變參數
redis.call('ZREM', zset_key, unpack(list))
local result = redis.call('HMGET', hash_key, unpack(list))
redis.call('HDEL', hash_key, unpack(list))
return result
end
end
end
returnnil


如果最小的分數小於等於當前時間戳，就將該任務取出來執行，否則休眠一段時間後再查詢。

註意：這裏其實有一個效能隱患，命令 ZREVRANGEBYSCORE 的時間復雜度可以視為為O(N)，N是集合的元素個數，由於這裏把所有的訂單資訊都放進了同一個Sorted Set(ORDER_QUEUE)中，所以在一直有新增數據的時候， dequeue 指令碼的時間復雜度一直比較高，後續訂單量升高之後會此處一定會成為效能瓶頸，後面會給出解決的方案

這裏的出隊使用 Crontab 作為輪訓去查詢消費

業務核心程式碼

延遲佇列類 RedisDelayQueue.php

<?php
/**
 * @desc Redis 延遲任務佇列
 * @author Tinywan(ShaoBo Wan)
 * @date 2024/05/02 11:36
 */
declare(strict_types=1);
namespaceredis;
classRedisDelayQueue
{
// 生產者 指令碼sha值
const DELAY_QUEUE_PRODUCER_SCRIPT_SHA = 'DELAY:QUEUE:PRODUCER:SCRIPT:SHA';
// 消費者 指令碼sha值
const DELAY_QUEUE_CONSUMER_SCRIPT_SHA = 'DELAY:QUEUE:CONSUMER:SCRIPT:SHA';
// 訂單關閉
const DELAY_QUEUE_ORDER_CLOSE = 'DELAY:QUEUE:ORDER:CLOSE';
// 訂單關閉詳情哈希
const DELAY_QUEUE_ORDER_CLOSE_HASH = 'DELAY:QUEUE:ORDER:CLOSE:HASH';
/**
* Redis 靜態例項
* @return \Redis
*/
privatestaticfunction_redis()
{
$redis = \redis\BaseRedis::server();
$redis->select(3);
return $redis;
}
/**
* @desc: 延遲佇列 生產者
* @param string $keys1
* @param string $keys2
* @param string $member
* @param int $score
* @param array $message
* @return mixed
*/
publicstaticfunctionproducer(string $keys1, string $keys2, string $member, int $score, array $message)
{
$redis = self::_redis();
$scriptSha = $redis->get(self::DELAY_QUEUE_PRODUCER_SCRIPT_SHA);
if (!$scriptSha) {
$script = <<<luascript
redis.call('ZADD', KEYS[1], ARGV[1], ARGV[2])
redis.call('HSET', KEYS[2], ARGV[2], ARGV[3])
return 1
luascript;
$scriptSha = $redis->script('load', $script);
$redis->set(self::DELAY_QUEUE_PRODUCER_SCRIPT_SHA, $scriptSha);
}
$hashValue = json_encode($message, JSON_UNESCAPED_UNICODE);
return $redis->evalSha($scriptSha, [$keys1, $keys2, $score, $member, $hashValue], 2);
}
/**
* @desc: 延遲佇列 消費者
* @param string $keys1
* @param string $keys2
* @param int $maxScore
* @return mixed
*/
publicstaticfunctionconsumer(string $keys1, string $keys2, int $maxScore)
{
$redis = self::_redis();
$scriptSha = $redis->get(self::DELAY_QUEUE_CONSUMER_SCRIPT_SHA);
if (!$scriptSha) {
$script = <<<luascript
local status, type = next(redis.call('TYPE', KEYS[1]))
if status ~= nil and status == 'ok' then
if type == 'zset' then
local list = redis.call('ZREVRANGEBYSCORE', KEYS[1], ARGV[1], ARGV[2], 'LIMIT', ARGV[3], ARGV[4])
if list ~= nil and #list > 0 then
redis.call('ZREM', KEYS[1], unpack(list))
local result = redis.call('HMGET', KEYS[2], unpack(list))
redis.call('HDEL', KEYS[2], unpack(list))
return result
end
end
end
luascript;
$scriptSha = $redis->script('load', $script);
$redis->set(self::DELAY_QUEUE_CONSUMER_SCRIPT_SHA, $scriptSha);
}
return $redis->evalSha($scriptSha, [$keys1, $keys2, $maxScore, 0, 0, 10], 2);
}
}








用redis來實作可以依賴於redis自身的持久化來實作持久化，redis的集群來支持高並行和高可用。因此開發成本很小，可以做到很即時。

指令碼命令列

生產者訊息

privatefunctiondelayQueueOrderClose()
{
$orderId = time();
$keys1 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE;
$keys2 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE_HASH;
$score = time() + 60; // 延遲60秒執行
$message = [
'event' => RedisDelayQueue::EVENT_ORDER_CLOSE,
'order_id' => $orderId,
'create_time' => time()
];
$res = RedisDelayQueue::producer($keys1, $keys2, (string) $orderId, $score, $message);
var_dump($res);
}


如果是ThinkPHP6 框架，執行該命令則可以生產訊息， php think crontab delay-queue-order-producer

迴圈

privatefunctiondelayOrderProducer()
{
$keys1 = DelayQueue::KEY_ORDER_CLOSE;
$keys2 = DelayQueue::KEY_ORDER_CLOSE_HASH;
for ($i = 1; $i <= 10; $i++) {
$orderId = 'S' . $i;
$score = time(); // 延遲60秒執行
$message = [
'event' => DelayQueue::EVENT_ORDER_CLOSE,
'order_id' => $orderId,
'create_time' => time()
];
$res = DelayQueue::producer($keys1, $keys2, (string) $orderId, $score, $message);
var_dump($res);
}
}


消費者訊息

1、透過Crontab 輪詢執行

privatefunctiondelayQueueOrderConsumer()
{
$keys1 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE;
$keys2 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE_HASH;
$maxScore = time();
$queueList = RedisDelayQueue::consumer($keys1, $keys2, $maxScore);
if (false === $queueList) {
echo' [x] Message List is Empty, Try Again ', "\n";
return;
}
var_dump($queueList);
}


說明：如果最小的分數小於等於當前時間戳，就將該任務取出來執行，否則休眠一段時間後再查詢

2、阻塞執行

privatefunctiondelayQueueOrderConsumerWhile()
{
$keys1 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE;
$keys2 = RedisDelayQueue::DELAY_QUEUE_ORDER_CLOSE_HASH;
while (true) {
$maxScore = time();
$queueList = RedisDelayQueue::consumer($keys1, $keys2, $maxScore);
if (false === $queueList) {
echo' [x] Message List is Empty, Try Again ', "\n";
sleep(1);
continue;
}
// 處理業務
foreach ($queueList as $queue) {
$messageArray = json_decode($queue, true);
}
}
}


數據刪除為處理問題

方案一：彈出訂單內容數據的同時進行數據刪除，也就是 ZREVRANGEBYSCORE 、 ZREM 和 HDEL 命令要在同一個 Lua 指令碼中執行，這樣的話 Lua 指令碼的編寫難度大，並且由於彈出數據已經在Redis中刪除，如果數據處理失敗則可能需要從資料庫重新查詢補償。

針對以上的解決方案就是： 訊息進入到延遲佇列後，保證至少被消費一次。