需求

《Bug：Zabbix對(duì)Kafka topic積壓數(shù)據(jù)監(jiān)控》一文我們通過監(jiān)控lag來對(duì)Kafka某個(gè)分區(qū)topic的消費(fèi)情況進(jìn)行告警。通過告警我們發(fā)現(xiàn)，分區(qū)topic的消費(fèi)積壓情況告警非常頻繁，這無疑會(huì)引起開發(fā)、運(yùn)維的重點(diǎn)關(guān)注。經(jīng)過一系列的監(jiān)控、摸索、實(shí)踐、總結(jié)，我們逐步發(fā)現(xiàn)分區(qū)topic的消費(fèi)積壓和以下幾種情況有關(guān)：

• 消費(fèi)者組頻繁出現(xiàn)Rebalance，導(dǎo)致整個(gè)消費(fèi)者組下的topic都無法消費(fèi)；
• 消費(fèi)者性能問題，無法在超時(shí)時(shí)間內(nèi)完成消費(fèi)；
• topic分區(qū)數(shù)和消費(fèi)者數(shù)量不均衡，一個(gè)消費(fèi)者需要消費(fèi)多個(gè)分區(qū)topic，消費(fèi)緩慢；
• topic分區(qū)數(shù)量變化；
• 等等

從以上幾種情況分析，無論哪種都和消費(fèi)者組Rebalance有相關(guān)性，都是在經(jīng)過Rebalance后再重新消費(fèi)。因此我們還得從Rebalance的角度再出發(fā)。

Rebalance再出發(fā)

其中關(guān)于消費(fèi)者性能問題，這大多和客戶端的參數(shù)設(shè)置不恰當(dāng)相關(guān)，這是運(yùn)維比較難覺察導(dǎo)致。但是為了更全面的了解Kafka，我們運(yùn)維還是很有必要去輕了解下的。先從相關(guān)參數(shù)說起：

# 消費(fèi)者每次poll()最大消費(fèi)消息數(shù)量，默認(rèn)500條
max.poll.records
# 兩次poll()之間的最大間隔，默認(rèn)值為5分鐘
max.poll.interval.ms

• max.poll.interval.ms

這個(gè)參數(shù)定義了兩次poll()之間的最大間隔，「默認(rèn)值為5分鐘」。如果業(yè)務(wù)處理消息時(shí)間過長(zhǎng)，則會(huì)導(dǎo)致兩次poll()的時(shí)間間隔大于超時(shí)時(shí)間，從而觸發(fā)Rebalance。因此我們應(yīng)該適當(dāng)調(diào)整每次poll()的數(shù)量，以保證在規(guī)定時(shí)間內(nèi)處理完消息，這就需要關(guān)注max.poll.records參數(shù)了。

• max.poll.records

這個(gè)參數(shù)定義了poll()方法最多可以返回多少條消息，「默認(rèn)值為500」。poll()的數(shù)量如何定義，需要根據(jù)業(yè)務(wù)處理邏輯來決定，例如數(shù)據(jù)要經(jīng)過多個(gè)數(shù)據(jù)源進(jìn)行處理，一旦某一數(shù)據(jù)源訪問超時(shí)，無疑都會(huì)降低消費(fèi)效率。比較友好的解決方案是，開發(fā)可以根據(jù)不同的情況實(shí)時(shí)調(diào)整相關(guān)參數(shù)，應(yīng)用側(cè)動(dòng)態(tài)感知并進(jìn)行自動(dòng)熱加載，達(dá)到快速調(diào)整消費(fèi)的效果。

• 消費(fèi)者組劃分

除了對(duì)Kafka參數(shù)的調(diào)整，我們還要根據(jù)業(yè)務(wù)處理邏輯對(duì)消費(fèi)者組進(jìn)行提前規(guī)劃，避免為了方便將業(yè)務(wù)相關(guān)的topic同時(shí)劃分到同一個(gè)大消費(fèi)者組，這樣一旦某個(gè)消費(fèi)者出現(xiàn)問題，將會(huì)導(dǎo)致整個(gè)消費(fèi)者組重新Rebalace。如果Rebalance時(shí)間過長(zhǎng)，此時(shí)所有的topic無法消費(fèi)，那么實(shí)時(shí)業(yè)務(wù)將會(huì)受到很大的影響。因此我們要合理分配topic到不同的消費(fèi)者組。

監(jiān)控

經(jīng)過以上的探索分析，我們的首要任務(wù)應(yīng)該是監(jiān)控Kafka消費(fèi)者組是否處于Rebalance狀態(tài)，進(jìn)而確定：

分區(qū)消費(fèi)者是否發(fā)生切換，此時(shí)消費(fèi)者數(shù)量不變；

分區(qū)消費(fèi)者數(shù)量是否減少，出現(xiàn)一個(gè)消費(fèi)者同時(shí)消費(fèi)多個(gè)分區(qū)topic；

分區(qū)數(shù)量和消費(fèi)者是否為1:1關(guān)系，避免出現(xiàn)消費(fèi)者和分區(qū)數(shù)量不一致的情況；

1.監(jiān)控思路

在多消費(fèi)者組的情況下，我們不僅要監(jiān)控每個(gè)消費(fèi)者組的Rebalance的狀態(tài)，還要考慮到未來消費(fèi)者組的擴(kuò)展，因此我們希望可以通過配置文件的形式對(duì)消費(fèi)者進(jìn)行自動(dòng)發(fā)現(xiàn)并添加監(jiān)控。在此我們?nèi)匀皇峭ㄟ^Zabbix自動(dòng)發(fā)現(xiàn)，實(shí)現(xiàn)對(duì)每個(gè)消費(fèi)者組的Rebalance狀態(tài)進(jìn)行監(jiān)控告警。

2.消費(fèi)者組自動(dòng)發(fā)現(xiàn)

由于此配置文件和Kafk topic監(jiān)控復(fù)用同一個(gè)文件，通過Zabbix可對(duì)特定消費(fèi)者組(Group)進(jìn)行去重識(shí)別并行自動(dòng)發(fā)現(xiàn)。

# 自動(dòng)發(fā)現(xiàn)
vim consumer-groups.conf
#按消費(fèi)者組(Group)|Topic格式，寫入自動(dòng)發(fā)現(xiàn)配置文件
test-group|test

# 執(zhí)行腳本自動(dòng)發(fā)現(xiàn)所有的消費(fèi)者分組
bash consumer-groups-rebalance.sh discovery
{
    "data": [
        { "{#GROUP}":"test-group" }
    ]
}

3.獲取消費(fèi)者組Rebalance狀態(tài)

# 獲取rebalance狀態(tài)，0代表沒有rebalance，1代表處于rebalance
[root#~] bash consumer-groups-rebalance.sh status test-group
0

4.最終腳本

#!/bin/bash
#comment: 查詢消費(fèi)者組狀態(tài)，如果出現(xiàn)reabalance則進(jìn)行告警
#配置文件說明，和topic lag監(jiān)控共用一套配置文件
#消費(fèi)者組|Topic
#test-group|test

#加載環(huán)境變量
export JAVA_HOME=/usr/local/jdk1.8.0_261
export JRE_HOME=$JAVA_HOME/jre
export PATH=$JAVA_HOME/bin:$JRE_HOME/bin:$PATH

#group自動(dòng)發(fā)現(xiàn)
group_discovery() {
    printf "{\n"
    printf "\t\"data\": [\n"
    m=0
    num=`cat /etc/zabbix/monitor_scripts/consumer-groups.conf |awk -F'|' '{print $1}'|sort|uniq|wc -l`
    for group in `cat /etc/zabbix/monitor_scripts/consumer-groups.conf|awk -F'|' '{print $1}'|sort|uniq`
    do  
        m=`expr $m + 1`
        #判斷最后一行
        if  [ $m -eq $num ]; then
            printf "\t\t{ \"{#GROUP}\":\"${group}\" }\n"
        else
            printf "\t\t{ \"{#GROUP}\":\"${group}\" },\n"
        fi
    done
    printf "\t]\n"
    printf "}\n"
}

if [ $1 == "discovery" ]; then
    group_discovery
elif [ $1 == "status" ];then
    /usr/local/kafka/bin/./kafka-consumer-groups.sh --bootstrap-server 10.10.10.233:9092 --describe --group $2 > /tmp/$2 2>&1 
    cat /tmp/$2 |grep rebalanc |wc -l
else
    echo "Usage: /etc/zabbix/monitor_scripts/consumer-group.sh discovery | lag"
fi

5.Zabbix自動(dòng)發(fā)現(xiàn)

• 自動(dòng)發(fā)現(xiàn)配置

• 監(jiān)控項(xiàng)原型 以消費(fèi)者組定義監(jiān)控項(xiàng)名稱，告警信息中的名稱能夠幫助我們快速定位配置。

• 觸發(fā)器配置 告警觸發(fā)時(shí)，能夠通過告警信息快速定位kafka 消費(fèi)者組故障。

• 告警信息

告警主機(jī)：Kafka_192.168.3.55
主機(jī)IP：192.168.3.55
主機(jī)組：Kafka
告警時(shí)間：2022.11.11 11:22:00
恢復(fù)時(shí)間：2022.11.11 11:23:02
告警等級(jí)：Warning
告警信息：消費(fèi)者組test-group:處于rebalance狀態(tài)
告警項(xiàng)目：group_status[test-group]
問題詳情：
blaze-route: 1

其他運(yùn)維問題簡(jiǎn)單處理

# 1.手動(dòng)消費(fèi)某topic積壓的消息
/usr/local/kafka/bin/kafka-consumer-groups.sh --bootstrap-server 10.10.10.233:9092 --topic test --group test-group

# 2.調(diào)整kafka某個(gè)topic的數(shù)據(jù)有效期，有效釋放磁盤空間
/usr/local/kafka/bin/kafka-topics.sh -zookeeper 10.10.10.233:9092 --topic test-group --alter --config retention.ms=79200000

# 3.調(diào)整kafka某個(gè)topic的分區(qū)數(shù)量
/usr/local/kafka/bin/kafka-topics.sh -zookeeper 10.10.10.233:9092 --topic test-group --alter --partitions 4

以上是在使用Kafka過程中比較常用的幾個(gè)命令使用方式。