一、業(yè)務(wù)背景

從技術(shù)的角度來(lái)說(shuō)，技術(shù)方案的選型都是受限于實(shí)際的業(yè)務(wù)場(chǎng)景，都以解決實(shí)際業(yè)務(wù)場(chǎng)景為目標(biāo)。

在我們的實(shí)際業(yè)務(wù)場(chǎng)景中，需要以游戲的維度收集和上報(bào)行為數(shù)據(jù)，考慮數(shù)據(jù)的量級(jí)，執(zhí)行盡最大努力交付且允許數(shù)據(jù)的部分丟棄。

數(shù)據(jù)上報(bào)支持游戲的維度的批量上報(bào)，支持同一款游戲128個(gè)行為進(jìn)行批量上報(bào)。

數(shù)據(jù)上報(bào)需要時(shí)效控制，上報(bào)的數(shù)據(jù)必須是上報(bào)時(shí)刻的前3分鐘的數(shù)據(jù)。

整體數(shù)據(jù)的業(yè)務(wù)形態(tài)如下圖所示：

二、技術(shù)選型

從業(yè)務(wù)的角度來(lái)說(shuō)包含數(shù)據(jù)的收集和數(shù)據(jù)的上報(bào)，我們把數(shù)據(jù)的收集比作生產(chǎn)者，數(shù)據(jù)的上報(bào)比作消費(fèi)者，是一個(gè)典型的生產(chǎn)消費(fèi)模型。

生產(chǎn)消費(fèi)模型在JVM進(jìn)程內(nèi)部通過(guò)隊(duì)列+鎖或者無(wú)鎖的Disruptor來(lái)實(shí)現(xiàn)，在跨進(jìn)程場(chǎng)景下通過(guò)MQ（RocketMQ/kafka）進(jìn)行處理解耦。

但是細(xì)化到具體業(yè)務(wù)場(chǎng)景來(lái)看，消息的消費(fèi)有諸多限制，包括： 游戲維度的批量行為上報(bào)，行為上報(bào)的時(shí)效限制，細(xì)化到各個(gè)技術(shù)方案選型進(jìn)行對(duì)比。

方案一

使用RocketMQ 或者Kafaka等消息隊(duì)列來(lái)存儲(chǔ)上報(bào)的消息，但是消費(fèi)側(cè)需要考慮在業(yè)務(wù)進(jìn)程中按照游戲維度進(jìn)行聚合，其中技術(shù)細(xì)節(jié)涉及按照游戲維度進(jìn)行拆分，在滿足消息時(shí)效性和批量性的前提下觸發(fā)上報(bào)。在這種方案下消息中間件扮演的角色本質(zhì)上消息的中轉(zhuǎn)站， 沒(méi)有解決任何業(yè)務(wù)場(chǎng)景中提及的游戲維度拆分、批量性和時(shí)效性。

方案二

在方案一的基礎(chǔ)上，尋求一種技術(shù)方案來(lái)解決游戲維度的 消息分組、批量消費(fèi) 、時(shí)效性 。通過(guò)Redis的list結(jié)構(gòu)來(lái)實(shí)現(xiàn)隊(duì)列（進(jìn)一步要求實(shí)現(xiàn)定長(zhǎng)隊(duì)列）來(lái)解決游戲維度的消息分組；通過(guò)Redis的list支持的Lrange來(lái)實(shí)現(xiàn)批量消費(fèi)；通過(guò)業(yè)務(wù)側(cè)的多線程來(lái)解決時(shí)效問(wèn)題，針對(duì)高頻的游戲使用單獨(dú)的線程池進(jìn)行處理，上述兩個(gè)手段能夠保證消費(fèi)速度大于生產(chǎn)速度。

方案對(duì)比

對(duì)比兩種方案后決定使用Redis的實(shí)現(xiàn)了一個(gè)偽消息中間件：

1. 通過(guò)List對(duì)象實(shí)現(xiàn)定長(zhǎng)隊(duì)列來(lái)保存游戲維度的行為消息（以游戲作為key的List對(duì)象來(lái)保存用戶行為）；
2. 通過(guò)List來(lái)保存所有的存在行為數(shù)據(jù)的游戲列表；
3. 通過(guò)Set來(lái)進(jìn)行去重判斷來(lái)保證2中的List對(duì)象的唯一性。

整體的技術(shù)方案如下圖所示：

生產(chǎn)過(guò)程

步驟一：游戲維度的某行為數(shù)據(jù)PUSH到游戲維度的隊(duì)列當(dāng)中。

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

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

消費(fèi)過(guò)程

步驟一：從游戲?qū)ο蟮牧斜碇醒h(huán)取出一款游戲。

步驟二：通過(guò)步驟一獲取的游戲?qū)ο笕ピ撚螒驅(qū)ο蟮男袨閿?shù)據(jù)隊(duì)列中批量獲取數(shù)據(jù)處理。

三、技術(shù)原理

在Redis的支持命令中，在List和Set的基礎(chǔ)命令，結(jié)合Lua腳本來(lái)實(shí)現(xiàn)整個(gè)技術(shù)方案。

消息數(shù)據(jù)層面，通過(guò)單獨(dú)的List循環(huán)維護(hù)待消費(fèi)的游戲維度的數(shù)據(jù)，每個(gè)游戲維度使用定長(zhǎng)的List來(lái)保存消息。

消息生產(chǎn)過(guò)程中，通過(guò)結(jié)合List的llen+lpop+rpush來(lái)實(shí)現(xiàn)游戲維度的定長(zhǎng)隊(duì)列，保證隊(duì)列的長(zhǎng)度可控。

消息消費(fèi)過(guò)程中，通過(guò)結(jié)合List的lrange+ltrim來(lái)實(shí)現(xiàn)游戲維度的消息的批量消費(fèi)。

在整個(gè)執(zhí)行的復(fù)雜度層面，需要保證時(shí)間復(fù)雜度在0(N)常量維度，保證時(shí)間可控。

3.1 Lua 腳本

EVAL script numkeys key [key ...] arg [arg ...]
    時(shí)間復(fù)雜度：取決于腳本本身的執(zhí)行的時(shí)間復(fù)雜度。

> 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 already completed.

Redis采用相同的Lua解釋器去運(yùn)行所有命令，我們可以保證，腳本的執(zhí)行是原子性的。作用就類似于加了MULTI/EXEC。

• Lua 腳本內(nèi)多個(gè)命令以原子性的方式執(zhí)行，保證了命令執(zhí)行的線程安全。
• Lua 腳本結(jié)合List命令實(shí)現(xiàn)定長(zhǎng)隊(duì)列，實(shí)現(xiàn)批量消費(fèi)。
• Lua 腳本僅支持單個(gè)key的操作，不支持多key的操作。

3.2 List 對(duì)象

LLEN key
    計(jì)算List的長(zhǎng)度
    時(shí)間復(fù)雜度：O(1)。

LPOP key [count]
    從List的左側(cè)移除元素
    時(shí)間復(fù)雜度：O(N)，N為移除元素的個(gè)數(shù)。
    
RPUSH key element [element ...]
    從List的右側(cè)保存元素
    時(shí)間復(fù)雜度：O(N)，N為保存元素的個(gè)數(shù)。

• List的基礎(chǔ)命令包括計(jì)算List的長(zhǎng)度，移除數(shù)據(jù)，添加數(shù)據(jù)，整體命令的復(fù)雜度都在O(N)的常量時(shí)間。
• 整合上述三個(gè)命令，我們能保證實(shí)現(xiàn)固定長(zhǎng)度的隊(duì)列，通過(guò)判斷隊(duì)列長(zhǎng)度是否達(dá)到定長(zhǎng)結(jié)合新增隊(duì)列元素和移除隊(duì)列元素來(lái)完成。

LRANGE key start end
    時(shí)間復(fù)雜度：O(S+N)， S為偏移量start， N為指定區(qū)間內(nèi)元素的數(shù)量。

    下標(biāo)(index)參數(shù) start 和 stop 都以 0 為底，也就是說(shuō)，以 0 表示列表的第一個(gè)元素，以 1 表示列表的第二個(gè)元素，以此類推。
    你也可以使用負(fù)數(shù)下標(biāo)，以 -1 表示列表的最后一個(gè)元素， -2 表示列表的倒數(shù)第二個(gè)元素，以此類推。

LTRIM key start stop
    時(shí)間復(fù)雜度：O(N) where N is the number of elements to be removed by the operation.

    修剪(trim)一個(gè)已存在的 list，這樣 list 就會(huì)只包含指定范圍的指定元素。

• List的基礎(chǔ)命令包括批量返回?cái)?shù)據(jù)和裁剪數(shù)據(jù)，整體命令的復(fù)雜度都在O(N)的常量時(shí)間。
• 整合上述兩個(gè)命令，我們能夠批量消費(fèi)數(shù)據(jù)并移除隊(duì)列數(shù)據(jù)，通過(guò)LRANGE批量返回?cái)?shù)據(jù)并通過(guò)LTRIM保留剩余數(shù)據(jù)。

3.3 Set 對(duì)象

SADD key member [member ...]
    往Set集合添加數(shù)據(jù)。

    時(shí)間復(fù)雜度：O(1)。

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

• 通過(guò)Set集合來(lái)保證數(shù)據(jù)的唯一性，且時(shí)間復(fù)雜度可控。

四、技術(shù)應(yīng)用

4.1 生產(chǎn)消息

定義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";

執(zhí)行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));

執(zhí)行效果
    實(shí)現(xiàn)固長(zhǎng)隊(duì)列的數(shù)據(jù)存儲(chǔ)并設(shè)置過(guò)期時(shí)間

• 通過(guò)整合llen+rpush+lpop三個(gè)命令實(shí)現(xiàn)定長(zhǎng)隊(duì)列。
• 通過(guò)lua腳本保證上述命令的原子性執(zhí)行。

• 整體的執(zhí)行流程如上圖所示，核心理念通過(guò)lua腳本的原子性保證了隊(duì)列長(zhǎng)度計(jì)算（llen）、隊(duì)列數(shù)據(jù)移除（lpop）、隊(duì)列數(shù)據(jù)保存（rpush)的原子性執(zhí)行。

4.2 消費(fèi)消息

定義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";

執(zhí)行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));

執(zhí)行效果
    取固定數(shù)量的對(duì)象，然后保留隊(duì)列的剩余的消息對(duì)象。

• 通過(guò)整合lrange+ltrim兩個(gè)命令實(shí)現(xiàn)消息的批量消費(fèi)。
• 通過(guò)lua腳本保證上述命令的原子性執(zhí)行。

• 整體的執(zhí)行流程如上圖所示，核心理念通過(guò)lua腳本的原子性保證了數(shù)據(jù)獲?。↙range）和數(shù)據(jù)裁剪（Ltrim)的原子性執(zhí)行。
• 整體的消費(fèi)流程選擇pull模式，通過(guò)多線程循環(huán)輪詢可消費(fèi)的隊(duì)列進(jìn)行消費(fèi)。與借助于redis的pub/sub的通知機(jī)制實(shí)現(xiàn)消費(fèi)流程的push模式相比，pull模式成本更低效果更佳。

4.3 注意事項(xiàng)

• Redis集群模式下，執(zhí)行Lua腳本建議傳單key，多key會(huì)報(bào)重定向錯(cuò)誤。
• 在不同的Redis版本下，Lua腳本針對(duì)null的返回值處理不同，參考官方文檔。
• 消費(fèi)者的消費(fèi)過(guò)程中通過(guò)循環(huán)遍歷游戲列表，然后根據(jù)游戲去獲取對(duì)應(yīng)的消息對(duì)象，但是不同的游戲?qū)?yīng)的熱度不同，所以在消費(fèi)端我們通過(guò)配置的方式為熱門游戲單獨(dú)開啟消費(fèi)線程進(jìn)行消費(fèi)，相當(dāng)于針對(duì)不同游戲配置不同優(yōu)先級(jí)的消費(fèi)者。

五、線上效果

• 生產(chǎn)和消費(fèi)的QPS約為1w qps左右，整體上報(bào)QPS通過(guò)批量上報(bào)后會(huì)遠(yuǎn)低于生產(chǎn)的消息生產(chǎn)和消費(fèi)的QPS。
• 整體數(shù)據(jù)的使用游戲包名作為key進(jìn)行存儲(chǔ)，性能上不存在熱點(diǎn)的問(wèn)題。

六、適用場(chǎng)景

在描述完方案的原理和實(shí)現(xiàn)細(xì)節(jié)之后，進(jìn)一步對(duì)適用的業(yè)務(wù)場(chǎng)景進(jìn)行下總結(jié)。整體方案是基于redis的基本數(shù)據(jù)結(jié)構(gòu)構(gòu)建一個(gè)偽消息隊(duì)列，用以解決 消息的單個(gè)生產(chǎn)批量消費(fèi) 的場(chǎng)景，通過(guò)多key形式實(shí)現(xiàn)消息隊(duì)列的多Topic模式，重要的是能夠借助于redis的原生能力在O(N)的時(shí)間復(fù)雜度完成批量消費(fèi)。另外該方案也可以降級(jí)作為實(shí)現(xiàn)先進(jìn)先出定長(zhǎng)的日志隊(duì)列。

七、總結(jié)

本文主要探索在特定業(yè)務(wù)場(chǎng)景下通過(guò)Redis的原生命令實(shí)現(xiàn)類MQ的功能，創(chuàng)新式的通過(guò)Lua腳本組合Redis的List的基礎(chǔ)命令，實(shí)現(xiàn)了消息的分組，消息的定長(zhǎng)隊(duì)列，消息的批量消費(fèi)功能；整體解決方案在線上環(huán)境落地并平穩(wěn)運(yùn)行，為特定場(chǎng)景提供了一種通用的解決方案。