Kafka消息積壓，你監(jiān)控Rebalance了嗎？

作者：三頁 2022-11-14 00:21:07

需求

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

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

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

Rebalance再出發(fā)

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

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

max.poll.interval.ms

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

max.poll.records

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

消費者組劃分

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

監(jiān)控

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

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

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

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

1.監(jiān)控思路

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

2.消費者組自動發(fā)現(xiàn)

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

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

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

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

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

4.最終腳本

#!/bin/bash
#comment: 查詢消費者組狀態(tài)，如果出現(xiàn)reabalance則進行告警
#配置文件說明，和topic lag監(jiān)控共用一套配置文件
#消費者組|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自動發(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自動發(fā)現(xiàn)

自動發(fā)現(xiàn)配置

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

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

告警信息

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

其他運維問題簡單處理

# 1.手動消費某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某個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某個topic的分區(qū)數(shù)量
/usr/local/kafka/bin/kafka-topics.sh -zookeeper 10.10.10.233:9092 --topic test-group --alter --partitions 4

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

責(zé)任編輯：武曉燕來源：木訥大叔愛運維

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看