kafka運維consumer-groups.sh消費者組管理

更新時間：2022年11月11日 10:32:25 作者：石臻臻的雜貨鋪

這篇文章主要為大家介紹了kafka運維consumer-groups.sh消費者組管理，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

消費者組管理 kafka-consumer-groups.sh

1. 查看消費者列表--list

sh bin/kafka-consumer-groups.sh --bootstrap-server xxxx:9090 --list

先調(diào)用MetadataRequest拿到所有在線Broker列表再給每個Broker發(fā)送ListGroupsRequest請求獲取消費者組數(shù)據(jù)

2. 查看消費者組詳情--describe

DescribeGroupsRequest

查看消費組詳情--group 或 --all-groups

查看指定消費組詳情--group sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --group test2_consumer_group

查看所有消費組詳情--all-groups sh bin/kafka-consumer-groups.sh --bootstrap-server xxxxx:9090 --describe --all-groups 查看該消費組消費的所有Topic、及所在分區(qū)、最新消費offset、Log最新數(shù)據(jù)offset、Lag還未消費數(shù)量、消費者ID等等信息

查詢消費者成員信息--members

所有消費組成員信息 sh bin/kafka-consumer-groups.sh --describe --all-groups --members --bootstrap-server xxx:9090

指定消費組成員信息 sh bin/kafka-consumer-groups.sh --describe --members --group test2_consumer_group --bootstrap-server xxxx:9090

查詢消費者狀態(tài)信息--state

所有消費組狀態(tài)信息 sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server xxxx:9090

指定消費組狀態(tài)信息 sh bin/kafka-consumer-groups.sh --describe --state --group test2_consumer_group --bootstrap-server xxxxx:9090

3. 刪除消費者組--delete

DeleteGroupsRequest

刪除消費組--delete

刪除指定消費組--group sh bin/kafka-consumer-groups.sh --delete --group test2_consumer_group --bootstrap-server xxxx:9090

刪除所有消費組--all-groups sh bin/kafka-consumer-groups.sh --delete --all-groups --bootstrap-server xxxx:9090

PS: 想要刪除消費組前提是這個消費組的所有客戶端都停止消費/不在線才能夠成功刪除;否則會報下面異常

Error: Deletion of some consumer groups failed:
* Group 'test2_consumer_group' could not be deleted due to: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.GroupNotEmptyException: The group is not empty.

4. 重置消費組的偏移量 --reset-offsets

能夠執(zhí)行成功的一個前提是消費組這會是不可用狀態(tài);

下面的示例使用的參數(shù)是: --dry-run ;這個參數(shù)表示預(yù)執(zhí)行,會打印出來將要處理的結(jié)果; 等你想真正執(zhí)行的時候請換成參數(shù)--excute ;

下面示例重置模式都是 --to-earliest 重置到最早的;

請根據(jù)需要參考下面相關(guān)重置Offset的模式換成其他模式;

重置指定消費組的偏移量 --group

重置指定消費組的所有Topic的偏移量--all-topic sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --all-topic

重置指定消費組的指定Topic的偏移量--topic sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --topic test2

重置所有消費組的偏移量 --all-group

重置所有消費組的所有Topic的偏移量--all-topic sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --all-topic

重置所有消費組中指定Topic的偏移量--topic sh bin/kafka-consumer-groups.sh --reset-offsets --to-earliest --all-group --bootstrap-server xxxx:9090 --dry-run --topic test2

--reset-offsets 后面需要接重置的模式

相關(guān)重置Offset的模式

參數(shù)	描述	例子
--to-earliest :	重置offset到最開始的那條offset(找到還未被刪除最早的那個offset)
--to-current:	直接重置offset到當前的offset，也就是LOE
--to-latest：	重置到最后一個offset
--to-datetime:	重置到指定時間的offset;格式為:YYYY-MM-DDTHH:mm:SS.sss;	--to-datetime "2021-6-26T00:00:00.000"
--to-offset	重置到指定的offset,但是通常情況下,匹配到多個分區(qū),這里是將匹配到的所有分區(qū)都重置到這一個值; 如果 1.目標最大offset<--to-offset, 這個時候重置為目標最大offset；2.目標最小offset>--to-offset ，則重置為最小; 3.否則的話才會重置為--to-offset的目標值; 一般不用這個	--to-offset 3465
--shift-by	按照偏移量增加或者減少多少個offset；正的為往前增加;負的往后退；當然這里也是匹配所有的;	--shift-by 100 、--shift-by -100
--from-file	根據(jù)CVS文檔來重置; 這里下面單獨講解

--to-offset 例子

--to-offset 3465

--from-file著重講解一下

上面其他的一些模式重置的都是匹配到的所有分區(qū); 不能夠每個分區(qū)重置到不同的offset；不過**--from-file**可以讓我們更靈活一點;

先配置cvs文檔格式為: Topic:分區(qū)號: 重置目標偏移量

test2,0,100
test2,1,200
test2,2,300

執(zhí)行命令

sh bin/kafka-consumer-groups.sh --reset-offsets --group test2_consumer_group --bootstrap-server xxxx:9090 --dry-run --from-file config/reset-offset.csv

5. 刪除偏移量delete-offsets

能夠執(zhí)行成功的一個前提是消費組這會是不可用狀態(tài);

偏移量被刪除了之后,Consumer Group下次啟動的時候,會從頭消費;

sh bin/kafka-consumer-groups.sh --delete-offsets --group test2_consumer_group2 --bootstrap-server XXXX:9090 --topic test2

相關(guān)可選參數(shù)

參數(shù)	描述	例子
--bootstrap-server	指定連接到的kafka服務(wù);	--bootstrap-server localhost:9092
--list	列出所有消費組名稱	--list
--describe	查詢消費者描述信息	--describe
--group	指定消費組
--all-groups	指定所有消費組
--members	查詢消費組的成員信息
--state	查詢消費者的狀態(tài)信息
--offsets	在查詢消費組描述信息的時候,這個參數(shù)會列出消息的偏移量信息; 默認就會有這個參數(shù)的;
dry-run	重置偏移量的時候,使用這個參數(shù)可以讓你預(yù)先看到重置情況，這個時候還沒有真正的執(zhí)行,真正執(zhí)行換成--excute;默認為dry-run
--excute	真正的執(zhí)行重置偏移量的操作;
--to-earliest	將offset重置到最早
to-latest	將offset重置到最近