Redis 定長佇列探索及實踐

2022-08-09 18:02:24

一、業務背景

從技術的角度來說，技術方案的選型都是受限於實際的業務場景，都以解決實際業務場景為目標。

在我們的實際業務場景中，需要以遊戲的維度收集和上報行為資料，考慮資料的量級，執行盡最大努力交付且允許資料的部分丟棄。

資料上報支援遊戲的維度的批次上報，支援同一款遊戲128個行為進行批次上報。

資料上報需要時效控制，上報的資料必須是上報時刻的前3分鐘的資料。

整體資料的業務形態如下圖所示：

二、技術選型

從業務的角度來說包含資料的收集和資料的上報，我們把資料的收集比作生產者，資料的上報比作消費者，是一個典型的生產消費模型。

生產消費模型在JVM程序內部通過佇列+鎖或者無鎖的Disruptor來實現，在跨程序場景下通過MQ（RocketMQ/kafka）進行處理解耦。

但是細化到具體業務場景來看，訊息的消費有諸多限制，包括：遊戲維度的批次行為上報，行為上報的時效限制，細化到各個技術方案選型進行對比。

方案一

使用RocketMQ 或者Kafaka等訊息佇列來儲存上報的訊息，但是消費側需要考慮在業務程序中按照遊戲維度進行聚合，其中技術細節涉及按照遊戲維度進行拆分，在滿足訊息時效性和批次性的前提下觸發上報。在這種方案下訊息中介軟體扮演的角色本質上訊息的中轉站，沒有解決任何業務場景中提及的遊戲維度拆分、批次性和時效性。

方案二

在方案一的基礎上，尋求一種技術方案來解決遊戲維度的訊息分組、批次消費、時效性。通過Redis的list結構來實現佇列（進一步要求實現定長佇列）來解決遊戲維度的訊息分組；通過Redis的list支援的Lrange來實現批次消費；通過業務側的多執行緒來解決時效問題，針對高頻的遊戲使用單獨的執行緒池進行處理，上述兩個手段能夠保證消費速度大於生產速度。

方案對比:

對比兩種方案後決定使用Redis的實現了一個偽訊息中介軟體：

通過List物件實現定長佇列來儲存遊戲維度的行為訊息（以遊戲作為key的List物件來儲存使用者行為）；

通過List來儲存所有的存在行為資料的遊戲列表；

通過Set來進行去重判斷來保證2中的List物件的唯一性。

整體的技術方案如下圖所示：

生產過程

步驟一：遊戲維度的某行為資料PUSH到遊戲維度的佇列當中。

步驟二：判斷遊戲是否在遊戲的集合Set中，如果在就直接返回，如果不在進行步驟三。

步驟三：往遊戲列表中PUSH遊戲。

消費過程

步驟一：從遊戲物件的列表中迴圈取出一款遊戲。

步驟二：通過步驟一獲取的遊戲物件去該遊戲物件的行為資料佇列中批次獲取資料處理。

三、技術原理

在Redis的支援命令中，在List和Set的基礎命令，結合Lua指令碼來實現整個技術方案。

訊息資料層面，通過單獨的List迴圈維護待消費的遊戲維度的資料，每個遊戲維度使用定長的List來儲存訊息。

訊息生產過程中，通過結合List的llen+lpop+rpush來實現遊戲維度的定長佇列，保證佇列的長度可控。

訊息消費過程中，通過結合List的lrange+ltrim來實現遊戲維度的訊息的批次消費。

在整個執行的複雜度層面，需要保證時間複雜度在0(N)常數維度，保證時間可控。

3.1 Lua 指令碼

EVAL script numkeys key [key ...] arg [arg ...]
時間複雜度：取決於指令碼本身的執行的時間複雜度。

> eval "return {KEYS[1],KEYS[2],ARGV[1],ARGV[2]}" 2 key1 key2 first second
1) "key1"
2) "key2"
3) "first"
4) "second"

Redis uses the same Lua interpreter to run all the commands.
Also Redis guarantees that a script is executed in an atomic way:
no other script or Redis command will be executed while a script is being executed.
This semantic is similar to the one of MULTI / EXEC.
From the point of view of all the other clients the effects of a script are either still not visible or

Redis採用相同的Lua直譯器去執行所有命令，我們可以保證，指令碼的執行是原子性的。作用就類似於加了MULTI/EXEC。

Lua 指令碼內多個命令以原子性的方式執行，保證了命令執行的執行緒安全。
Lua 指令碼結合List命令實現定長佇列，實現批次消費。
Lua 指令碼僅支援單個key的操作，不支援多key的操作。

3.2 List 物件

LLEN key
計算List的長度
時間複雜度：O(1)。

LPOP key [count]
從List的左側移除元素
時間複雜度：O(N)，N為移除元素的個數。

RPUSH key element [element ...]
從List的右側儲存元素
時間複雜度：O(N)，N為儲存元素的個數。

List的基礎命令包括計算List的長度，移除資料，新增資料，整體命令的複雜度都在O(N)的常數時間。
整合上述三個命令，我們能保證實現固定長度的佇列，通過判斷佇列長度是否達到定長結合新增佇列元素和移除佇列元素來完成。

LRANGE key start end
時間複雜度：O(S+N)， S為偏移量start， N為指定區間內元素的數量。

下標(index)引數 start 和 stop 都以 0 為底，也就是說，以 0 表示列表的第一個元素，以 1 表示列表的第二個元素，以此類推。
你也可以使用負數下標，以 -1 表示列表的最後一個元素， -2 表示列表的倒數第二個元素，以此類推。

LTRIM key start stop
時間複雜度：O(N) where N is the number of elements to be removed by

List的基礎命令包括批次返回資料和裁剪資料，整體命令的複雜度都在O(N)的常數時間。
整合上述兩個命令，我們能夠批次消費資料並移除佇列資料，通過LRANGE批次返回資料並通過LTRIM保留剩餘資料。

3.3 Set 物件

SADD key member [member ...]
往Set集合新增資料。
時間複雜度：O(1)。

SISMEMBER key member
判斷Set集合是否存在元素。
時間複雜度：O(1)。

四、技術應用

4.1 生產訊息

/*定義LUA指令碼*/
CACHE_NPPA_EVENT_LUA =
"local retVal = 0 " +
"local key = KEYS[1] " +
"local num = tonumber(ARGV[1]) " +
"local val = ARGV[2] " +
"local expire = tonumber(ARGV[3]) " +
"if (redis.call('llen', key) < num) then redis.call('rpush', key, val) " +
"else redis.call('lpop', key) redis.call('rpush', key, val) retVal = 1 end " +
"redis.call('expire', key, expire) return retVal";

/*執行LUA指令碼*/
String data = JSON.toJSONString(nppaBehavior);
Long retVal = (Long)jedisClusterTemplate.eval(CACHE_NPPA_EVENT_LUA, 1, NPPA_PREFIX + nppaBehavior.getGamePackage(), String.valueOf(MAX_GAME_EVENT_PER_GAME), data, String.valueOf(NPPA_TTL_MINUTE * 60));

/*執行效果
實現固長佇列的資料儲存並設定過期時間

通過整合llen+rpush+lpop三個命令實現定長佇列。
通過lua指令碼保證上述命令的原子性執行。

整體的執行流程如上圖所示，核心理念通過lua指令碼的原子性保證了佇列長度計算（llen）、佇列資料移除（lpop）、佇列資料儲存（rpush)的原子性執行。

4.2 消費訊息

/*定義LUA指令碼*/
QUERY_NPPA_EVENT_LUA =
"local data = {} " +
"local key = KEYS[1] " +
"local num = tonumber(ARGV[1]) " +
"data = redis.call('lrange', key, 0, num) redis.call('ltrim', key, num+1, -1) return data";

/*執行LUA指令碼*/
Integer batchSize = NppaConfigUtils.getInteger("nppa.report.batch.size", 1);
Object result = jedisClusterTemplate.eval(QUERY_NPPA_EVENT_LUA, 1,NPPA_PREFIX + gamePackage, String.valueOf(batchSize));

/*執行效果
取固定數量的物件，然後保留佇列的剩餘的訊息物件。

通過整合lrange+ltrim兩個命令實現訊息的批次消費。
通過lua指令碼保證上述命令的原子性執行。

整體的執行流程如上圖所示，核心理念通過lua指令碼的原子性保證了資料獲取（Lrange）和資料裁剪（Ltrim)的原子性執行。
整體的消費流程選擇pull模式，通過多執行緒迴圈輪詢可消費的佇列進行消費。與藉助於redis的pub/sub的通知機制實現消費流程的push模式相比，pull模式成本更低效果更佳。

4.3 注意事項

Redis叢集模式下，執行Lua指令碼建議傳單key，多key會報重定向錯誤。
在不同的Redis版本下，Lua指令碼針對null的返回值處理不同，參考官方檔案。
消費者的消費過程中通過迴圈遍歷遊戲列表，然後根據遊戲去獲取對應的訊息物件，但是不同的遊戲對應的熱度不同，所以在消費端我們通過設定的方式為熱門遊戲單獨開啟消費執行緒進行消費，相當於針對不同遊戲設定不同優先順序的消費者。

五、線上效果

生產和消費的QPS約為1w qps左右，整體上報QPS通過批次上報後會遠低於生產的訊息生產和消費的QPS。
整體資料的使用遊戲包名作為key進行儲存，效能上不存在熱點的問題。

六、適用場景

在描述完方案的原理和實現細節之後，進一步對適用的業務場景進行下總結。整體方案是基於redis的基本資料結構構建一個偽訊息佇列，用以解決訊息的單個生產批次消費的場景，通過多key形式實現訊息佇列的多Topic模式，重要的是能夠藉助於redis的原生能力在O(N)的時間複雜度完成批次消費。另外該方案也可以降級作為實現先進先出定長的紀錄檔佇列。