本文所有命令,博主均全部操作驗證過,保證準確性; 非複製貼上拼湊文章; 如果想了解更多工具命令,可在評論區留下評論,博主會擇期加上;

博主正在連載 Kafka原始碼、Kafka運維、Kafka實踐系列文章 並且相關文章會配套錄製視訊
本文為專欄第一篇歡迎關注<石臻臻的雜貨鋪>不迷路!!!

以下大部分運維操作,都可以使用 LogI-Kafka-Manager 在平臺上視覺化操作;

@

• 1.TopicCommand
  • 1.1.Topic建立
  • 1.2.刪除Topic
  • 1.3.Topic分割槽擴容
  • 1.4.查詢Topic描述
  • 5.查詢Topic列表
• 2.ConfigCommand
  • 2.1 查詢配置
    • Topic配置查詢
    • 其他配置/clients/users/brokers/broker-loggers 的查詢
    • 查詢kafka版本資訊
  • 2.2 增刪改 配置 --alter
    • Topic新增/修改動態配置
    • Topic刪除動態配置
    • 其他配置同理,只需要型別改下--entity-type
• 3.副本擴縮、分割槽遷移、跨路徑遷移 kafka-reassign-partitions
• 4.Topic的傳送kafka-console-producer.sh
• 5. Topic的消費kafka-console-consumer.sh
• 6.kafka-leader-election Leader重新選舉
• 7. 持續批量推送訊息kafka-verifiable-producer.sh
• 8. 持續批量拉取訊息kafka-verifiable-consumer
• 9.生產者壓力測試kafka-producer-perf-test.sh
• 10.消費者壓力測試kafka-consumer-perf-test.sh
• 11.刪除指定分割槽的訊息kafka-delete-records.sh
• 12. 檢視Broker磁碟資訊
• 12. 消費者組管理 kafka-consumer-groups.sh
  • 1. 檢視消費者列表--list
  • 2. 檢視消費者組詳情--describe
  • 3. 刪除消費者組--delete
  • 4. 重置消費組的偏移量 --reset-offsets
  • 5. 刪除偏移量delete-offsets
• 附件

1.TopicCommand

1.1.Topic建立

bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 3 --partitions 3 --topic test

相關可選引數

1.2.刪除Topic

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic test

支援正則表示式匹配Topic來進行刪除,只需要將topic 用雙引號包裹起來
例如: 刪除以create_topic_byhand_zk為開頭的topic;

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic "create_topic_byhand_zk.*"
.表示任意匹配除換行符 \n 之外的任何單字元。要匹配 . ，請使用 . 。
·*·：匹配前面的子表示式零次或多次。要匹配 * 字元，請使用 *。
.* : 任意字元

刪除任意Topic (慎用)

bin/kafka-topics.sh --bootstrap-server localhost:9092 --delete --topic ".*?"

更多的用法請參考正則表示式

1.3.Topic分割槽擴容

zk方式(不推薦)

>bin/kafka-topics.sh --zookeeper localhost:2181 --alter --topic topic1 --partitions 2

kafka版本 >= 2.2 支援下面方式（推薦）

單個Topic擴容

bin/kafka-topics.sh --bootstrap-server broker_host:port --alter --topic test_create_topic1 --partitions 4

批量擴容 (將所有正則表示式匹配到的Topic分割槽擴容到4個)

sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server 172.23.248.85:9092 --alter --partitions 4

".*?" 正則表示式的意思是匹配所有; 您可按需匹配

PS: 當某個Topic的分割槽少於指定的分割槽數時候,他會丟擲異常;但是不會影響其他Topic正常進行;

相關可選引數

PS: 雖然這裡配置的是全部的分割槽副本分配配置,但是正在生效的是新增的分割槽;
比如: 以前3分割槽1副本是這樣的

現在新增一個分割槽,--replica-assignment 2,1,3,4 ; 看這個意思好像是把0，1號分割槽互相換個Broker

但是實際上不會這樣做,Controller在處理的時候會把前面3個截掉; 只取新增的分割槽分配方式,原來的還是不會變

1.4.查詢Topic描述

1.查詢單個Topic

sh bin/kafka-topics.sh --topic test --bootstrap-server xxxx:9092 --describe --exclude-internal

2.批量查詢Topic(正則表示式匹配,下面是查詢所有Topic)

sh bin/kafka-topics.sh --topic ".*?" --bootstrap-server xxxx:9092 --describe --exclude-internal

支援正則表示式匹配Topic,只需要將topic 用雙引號包裹起來

相關可選引數

5.查詢Topic列表

1.查詢所有Topic列表

sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal

2.查詢匹配Topic列表(正則表示式)

查詢test_create_開頭的所有Topic列表
sh bin/kafka-topics.sh --bootstrap-server xxxxxx:9092 --list --exclude-internal --topic "test_create_.*"

相關可選引數

2.ConfigCommand

Config相關操作; 動態配置可以覆蓋預設的靜態配置;

2.1 查詢配置

Topic配置查詢

展示關於Topic的動靜態配置

1.查詢單個Topic配置(只列舉動態配置)

sh bin/kafka-configs.sh --describe --bootstrap-server xxxxx:9092 --topic test_create_topic
或者
sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics --entity-name test_create_topic

2.查詢所有Topic配置(包括內部Topic)(只列舉動態配置)

sh bin/kafka-configs.sh --describe --bootstrap-server 172.23.248.85:9092 --entity-type topics

3.查詢Topic的詳細配置(動態+靜態)

只需要加上一個引數--all

其他配置/clients/users/brokers/broker-loggers 的查詢

同理 ；只需要將--entity-type 改成對應的型別就行了 (topics/clients/users/brokers/broker-loggers)

查詢kafka版本資訊

sh bin/kafka-configs.sh --describe --bootstrap-server xxxx:9092 --version

所有可配置的動態配置 請看最後面的 附件 部分

2.2 增刪改 配置 --alter

--alter

刪除配置: --delete-config k1=v1,k2=v2
新增/修改配置: --add-config k1,k2
選擇型別: --entity-type (topics/clients/users/brokers/broker-
loggers)
型別名稱: --entity-name

Topic新增/修改動態配置

--add-config

sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --add-config file.delete.delay.ms=222222,retention.ms=999999

Topic刪除動態配置

--delete-config

sh bin/kafka-configs.sh --bootstrap-server xxxxx:9092 --alter --entity-type topics --entity-name test_create_topic1 --delete-config file.delete.delay.ms,retention.ms

其他配置同理,只需要型別改下--entity-type

型別有: (topics/clients/users/brokers/broker- loggers)

哪些配置可以修改 請看最後面的附件：ConfigCommand 的一些可選配置

3.副本擴縮、分割槽遷移、跨路徑遷移 kafka-reassign-partitions

請戳 【kafka運維】副本擴縮容、資料遷移、副本重分配、副本跨路徑遷移 (如果點不出來,表示文章暫未發表,請耐心等待)

4.Topic的傳送kafka-console-producer.sh

4.1 生產無key訊息

## 生產者
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties

4.2 生產有key訊息
加上屬性--property parse.key=true

## 生產者
bin/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic test --producer.config config/producer.properties  --property parse.key=true

預設訊息key與訊息value間使用“Tab鍵”進行分隔，所以訊息key以及value中切勿使用轉義字元(\t)

可選引數

5. Topic的消費kafka-console-consumer.sh

1. 新客戶端從頭消費--from-beginning (注意這裡是新客戶端,如果之前已經消費過了是不會從頭消費的)
下面沒有指定客戶端名稱,所以每次執行都是新客戶端都會從頭消費

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

2. 正則表示式匹配topic進行消費--whitelist
消費所有的topic

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist '.*'

消費所有的topic，並且還從頭消費

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --whitelist '.*' --from-beginning

3.顯示key進行消費--property print.key=true

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --property print.key=true

4. 指定分割槽消費--partition 指定起始偏移量消費--offset

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --partition 0 --offset 100

5. 給客戶端命名--group

注意給客戶端命名之後,如果之前有過消費，那麼--from-beginning 就不會再從頭消費了

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --group test-group

6. 新增客戶端屬性--consumer-property

這個引數也可以給客戶端新增屬性,但是注意 不能多個地方配置同一個屬性,他們是互斥的;比如在下面的基礎上還加上屬性--group test-group 那肯定不行

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer-property group.id=test-consumer-group

7. 新增客戶端屬性--consumer.config

跟--consumer-property 一樣的性質,都是新增客戶端的屬性,不過這裡是指定一個檔案,把屬性寫在檔案裡面, --consumer-property 的優先順序大於 --consumer.config

sh bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --consumer.config config/consumer.properties

6.kafka-leader-election Leader重新選舉

6.1 指定Topic指定分割槽用重新PREFERRED：優先副本策略 進行Leader重選舉

> sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --topic test_create_topic4 --election-type PREFERRED --partition 0

6.2 所有Topic所有分割槽用重新PREFERRED：優先副本策略 進行Leader重選舉

sh bin/kafka-leader-election.sh --bootstrap-server xxxx:9090 --election-type preferred  --all-topic-partitions

6.3 設定配置檔案批量指定topic和分割槽進行Leader重選舉

先配置leader-election.json檔案

{
  "partitions": [
    {
      "topic": "test_create_topic4",
      "partition": 1
    },
    {
      "topic": "test_create_topic4",
      "partition": 2
    }
  ]
}

 sh bin/kafka-leader-election.sh --bootstrap-server xxx:9090 --election-type preferred  --path-to-json-file config/leader-election.json
 

相關可選引數

7. 持續批量推送訊息kafka-verifiable-producer.sh

單次傳送100條訊息--max-messages 100

一共要推送多少條，預設為-1，-1表示一直推送到程序關閉位置

sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 --max-messages 100

每秒傳送最大吞吐量不超過訊息 --throughput 100

推送訊息時的吞吐量，單位messages/sec。預設為-1，表示沒有限制

sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 --throughput 100

傳送的訊息體帶字首--value-prefix

sh bin/kafka-verifiable-producer.sh --topic test_create_topic4 --bootstrap-server localhost:9092 --value-prefix 666

注意 --value-prefix 666必須是整數,傳送的訊息體的格式是加上一個 點號. 例如： 666.

其他引數：
--producer.config CONFIG_FILE 指定producer的配置檔案
--acks ACKS 每次推送訊息的ack值，預設是-1

8. 持續批量拉取訊息kafka-verifiable-consumer

持續消費

sh bin/kafka-verifiable-consumer.sh --group-id test_consumer --bootstrap-server localhost:9092 --topic test_create_topic4

單次最大消費10條訊息--max-messages 10

sh bin/kafka-verifiable-consumer.sh --group-id test_consumer --bootstrap-server localhost:9092 --topic test_create_topic4 --max-messages 10

相關可選引數

9.生產者壓力測試kafka-producer-perf-test.sh

1. 傳送1024條訊息--num-records 100並且每條訊息大小為1KB--record-size 1024 最大吞吐量每秒10000條--throughput 100

sh bin/kafka-producer-perf-test.sh --topic test_create_topic4 --num-records 100 --throughput 100000 --producer-props bootstrap.servers=localhost:9092 --record-size 1024

你可以通過LogIKM檢視分割槽是否增加了對應的資料大小

從LogIKM 可以看到傳送了1024條訊息; 並且總資料量=1M; 1024條*1024byte = 1M;

2. 用指定訊息檔案--payload-file 傳送100條訊息最大吞吐量每秒100條--throughput 100

1. 先配置好訊息檔案batchmessage.txt
   

2. 然後執行命令
   傳送的訊息會從batchmessage.txt裡面隨機選擇; 注意這裡我們沒有用引數--payload-delimeter指定分隔符，預設分隔符是\n換行;

   bin/kafka-producer-perf-test.sh --topic test_create_topic4 --num-records 100 --throughput 100 --producer-props bootstrap.servers=localhost:9090 --payload-file config/batchmessage.txt

3. 驗證訊息，可以通過 LogIKM 檢視傳送的訊息

相關可選引數

10.消費者壓力測試kafka-consumer-perf-test.sh

消費100條訊息 --messages 100

sh bin/kafka-consumer-perf-test.sh -topic test_create_topic4 --bootstrap-server localhost:9090 --messages 100

相關可選引數

11.刪除指定分割槽的訊息kafka-delete-records.sh

刪除指定topic的某個分割槽的訊息刪除至offset為1024

先配置json檔案offset-json-file.json

{"partitions":
[{"topic": "test1", "partition": 0,
  "offset": 1024}],
  "version":1
}

在執行命令

sh bin/kafka-delete-records.sh --bootstrap-server 172.23.250.249:9090 --offset-json-file config/offset-json-file.json

驗證 通過 LogIKM 檢視傳送的訊息

從這裡可以看出來,配置"offset": 1024 的意思是從最開始的地方刪除訊息到 1024的offset; 是從最前面開始刪除的

12. 檢視Broker磁碟資訊

查詢指定topic磁碟資訊 --topic-list topic1,topic2

sh bin/kafka-log-dirs.sh --bootstrap-server xxxx:9090 --describe --topic-list test2

查詢指定Broker磁碟資訊--broker-list 0 broker1,broker2

sh bin/kafka-log-dirs.sh --bootstrap-server xxxxx:9090 --describe --topic-list test2 --broker-list 0

例如我一個3分割槽3副本的Topic的查出來的資訊
logDir Broker中配置的log.dir

{
	"version": 1,
	"brokers": [{
		"broker": 0,
		"logDirs": [{
			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-0",
			"error": null,
			"partitions": [{
				"partition": "test2-1",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-0",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-2",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}]
		}]
	}, {
		"broker": 1,
		"logDirs": [{
			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-1",
			"error": null,
			"partitions": [{
				"partition": "test2-1",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-0",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-2",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}]
		}]
	}, {
		"broker": 2,
		"logDirs": [{
			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-2",
			"error": null,
			"partitions": [{
				"partition": "test2-1",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-0",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}, {
				"partition": "test2-2",
				"size": 0,
				"offsetLag": 0,
				"isFuture": false
			}]
		}]
	}, {
		"broker": 3,
		"logDirs": [{
			"logDir": "/Users/xxxx/work/IdeaPj/ss/kafka/kafka-logs-3",
			"error": null,
			"partitions": []
		}]
	}]
}

如果你覺得通過命令查詢磁碟資訊比較麻煩，你也可以通過 LogIKM 檢視

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

1. 檢視消費者列表--list

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

先呼叫MetadataRequest拿到所有線上Broker列表
再給每個Broker傳送ListGroupsRequest請求獲取 消費者組資料

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、及所在分割槽、最新消費offset、Log最新資料offset、Lag還未消費數量、消費者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

查詢消費者狀態資訊--state

所有消費組狀態資訊
sh bin/kafka-consumer-groups.sh --describe --all-groups --state --bootstrap-server xxxx:9090
指定消費組狀態資訊
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

能夠執行成功的一個前提是 消費組這會是不可用狀態;

下面的示例使用的引數是: --dry-run ;這個引數表示預執行,會打印出來將要處理的結果;
等你想真正執行的時候請換成引數--excute ;

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

請根據需要參考下面 相關重置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 後面需要接重置的模式

相關重置Offset的模式

--from-file著重講解一下

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

1. 先配置cvs文件
   格式為: Topic:分割槽號: 重置目標偏移量

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

2. 執行命令

   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

能夠執行成功的一個前提是 消費組這會是不可用狀態;

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

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

相關可選引數

附件

ConfigCommand 的一些可選配置

Topic相關可選配置

Broker相關可選配置

Users相關可選配置

clients相關可選配置

以上大部分運維操作,都可以使用 LogI-Kafka-Manager 在平臺上視覺化操作;

滴滴開源Logi-KafkaManager 一站式Kafka監控與管控平臺

歡迎加個人微信拉你進開發技術交流群,群內專人解答技術疑問
(請備註：技術) wx: jjdlmn_ 或 wx: mike_zhangliang