[[348392]]

本文轉(zhuǎn)載自微信公眾號「咖啡拿鐵」，作者咖啡拿鐵 。轉(zhuǎn)載本文請聯(lián)系咖啡拿鐵公眾號。

背景

今天在脈脈上面看到了一個(gè)帖子，比較有意思：

這個(gè)帖子的意思是：在使用Kafka的時(shí)候，我們已經(jīng)設(shè)置了多個(gè)分區(qū)，如何去提升消費(fèi)能力?如果使用線程池的方式去提升如何保證重啟時(shí)消息不丟。

這個(gè)題其實(shí)問了兩個(gè)點(diǎn)，第一個(gè)是如何提升消費(fèi)能力，第二個(gè)是如果選擇線程池，我們?nèi)绾巫龅较⒉粊G。

這里先解釋一下這兩個(gè)問題到底是怎么回事，在很多消息隊(duì)列中都有一個(gè)概念叫partion，代表著分區(qū)，分區(qū)是我們提高消息隊(duì)列消費(fèi)的關(guān)鍵，我們的消費(fèi)者消費(fèi)的渠道就是從每個(gè)分區(qū)中來的，一個(gè)分區(qū)只能被一個(gè)消費(fèi)者持有，如下圖所示：

有點(diǎn)類似銀行排隊(duì)，隊(duì)列的個(gè)數(shù)越多，排隊(duì)的時(shí)間相對來說就會越少，當(dāng)然也可以通過異步的方式去處理，比如線程池，把所有的消息都扔到線程池中去執(zhí)行，這就引出了作者說的第二個(gè)問題，首先我們來看看同步消費(fèi)為什么不會丟消息呢?

如果我們使用的是同步模型，當(dāng)我們消費(fèi)了之后會將offset ack回去，如果我們出現(xiàn)了重啟，沒有成功offset，那么這部分?jǐn)?shù)據(jù)將會再次消費(fèi)，如果是用線程池進(jìn)行消費(fèi)，那么我們?nèi)绾芜M(jìn)行ack呢，比如我們用線程池消費(fèi)了 10，11，12 三條消息如果12先消費(fèi)完，那么我們ack 13嗎?如果這樣做的話，這個(gè)時(shí)候重啟，kafka就會認(rèn)為你已經(jīng)處理了10，11的消息，這個(gè)時(shí)候消息就會出現(xiàn)丟失，而發(fā)這個(gè)帖子的同學(xué)就是對于這一塊是比較疑惑。

網(wǎng)友的回答

我們來看看網(wǎng)友的一些回答：

網(wǎng)友A:

這名網(wǎng)友的回答本質(zhì)還是使用線程池，作者也回復(fù)了，并沒有解決線程池的問題。

網(wǎng)友B:

這個(gè)方法類似銀行排隊(duì)，只要隊(duì)列多，那么處理速度就會加快，的確是第一個(gè)問題的解決辦法之一。

網(wǎng)友C:

這一類主要解決了第二個(gè)問題，通過外部維護(hù)offset,比如通過offset入庫的方式，我們就能找到正確的應(yīng)該消費(fèi)的offset，這個(gè)相對來說比較復(fù)雜，使用一個(gè)MQ還得配套一個(gè)數(shù)據(jù)庫，萬一我使用MQ的服務(wù)根本都沒有數(shù)據(jù)庫，還得單獨(dú)去申請。

網(wǎng)友D:

還有另外一種觀點(diǎn)就是，代碼寫好一點(diǎn)，讓消費(fèi)的速度提高，那消費(fèi)能力自然就上去了，這個(gè)的確是一個(gè)很重要的點(diǎn)，通常被其他人給忽略，有時(shí)候消費(fèi)比較慢，很多人可能一上來就是考慮中間件應(yīng)該怎么設(shè)置，往往會忽略自己的代碼。

看了這么多帖子的一個(gè)回復(fù)，感覺沒有真正能讓我滿意的答案，下面來說說我心中的一些思路。

我的想法

對于第一個(gè)問題的話，如何提升消費(fèi)能力?這個(gè)問題其實(shí)可以總結(jié)為三個(gè)辦法：

1. 如果每臺消費(fèi)者機(jī)器消費(fèi)線程是固定的，那么我們可以擴(kuò)容消費(fèi)機(jī)器和partion,類似銀行排隊(duì)增加排隊(duì)窗口一樣。
2. 如果機(jī)器和partion是固定的，增加消費(fèi)線程就是一個(gè)比較好的辦法，但是如果是順序消費(fèi)，就不能通過增加線程數(shù)的方式來提升消費(fèi)能力，因?yàn)轫樞蛳M(fèi)每個(gè)partion都是一個(gè)單獨(dú)的線程，只能通過第一種方式去解決。
3. 增加自身代碼的消費(fèi)能力，你想想如果銀行辦事，如果柜員的辦事效率能提升的非常高，那么整個(gè)排隊(duì)速度肯定也是很快的。

對于第二個(gè)問題，如果我們使用線程池模型,如何去解決消息丟失問題，這里我比較推薦的是RocketMQ中的做法，我們之前說了用數(shù)據(jù)庫去保存offset比較復(fù)雜，性能還比較差，在RocketMQ中使用了一個(gè)TreeMap的結(jié)構(gòu)做了我們上面提到的數(shù)據(jù)庫的事:

private final TreeMap<Long, MessageExt> msgTreeMap = new TreeMap<Long, MessageExt>(); 

這個(gè)TreeMap的key是每個(gè)message的offset，value就是這條消息的一些信息，TreeMap的底層是使用紅黑樹去實(shí)現(xiàn)的，我們可以很快獲取其中的最小值和最大值，當(dāng)我們每次處理完某一條消息的時(shí)候我們會將這條消息從msgTreeMap中移除，

public long removeMessage(final List<MessageExt> msgs) { 
        long result = -1; 
        final long now = System.currentTimeMillis(); 
        try { 
            this.lockTreeMap.writeLock().lockInterruptibly(); 
            this.lastConsumeTimestamp = now; 
            try { 
                if (!msgTreeMap.isEmpty()) { 
                    result = this.queueOffsetMax + 1; 
                    int removedCnt = 0; 
                    for (MessageExt msg : msgs) { 
                        MessageExt prev = msgTreeMap.remove(msg.getQueueOffset()); 
                        if (prev != null) { 
                            removedCnt--; 
                            msgSize.addAndGet(0 - msg.getBody().length); 
                        } 
                    } 
                    msgCount.addAndGet(removedCnt); 
 
                    if (!msgTreeMap.isEmpty()) { 
                        result = msgTreeMap.firstKey(); 
                    } 
                } 
            } finally { 
                this.lockTreeMap.writeLock().unlock(); 
            } 
        } catch (Throwable t) { 
            log.error("removeMessage exception", t); 
        } 
        return result; 
    } 

removeMessage這個(gè)方法就是移除已經(jīng)消費(fèi)過的消息，并且返回當(dāng)前最新的消費(fèi)offset，這里返回的結(jié)果就是msgTreeMap.firstKey()，我們ack給消息隊(duì)列server的值其實(shí)也是這個(gè)，回到我們這個(gè)問題上，如果我們發(fā)生重啟，那么其實(shí)也不需要擔(dān)心我們會出現(xiàn)消息丟失。