目錄

1、集群運維命令

1.1、集群啟停命令

1.3、集群遷移命令

1.4、權限管理命令

1.4.1、權限參數(shù)介紹

1.4.2、增加權限命令

1.4.3、移出權限命令

1.4.4、查看所有topic權限命令

1.4.5、查看某個topic權限命令

2、生產者命令

2.1、創(chuàng)建topic命令

2.2、刪除topic命令

2.3、修改topic命令

2.3.1、修改配置命令

2.3.2、刪除配置命令

2.4、查詢topic命令

2.4.1、查詢topic列表命令

2.4.2、查看某個topic詳情命令

?2.4.2、查看某個topic配置命令

2.5、發(fā)送消息命令

3、消費者命令

3.1、消費者組命令

3.1.1、查看消費者組列表命令

3.1.2、查看消費者組成員命令

3.1.3、查看某個消費組詳情命令

3.1.4、查看某個消費者組狀態(tài)命令

3.3、消費消息命令

4、性能測試命令

5、類執(zhí)行命令

---

1、集群運維命令

1.1、集群啟停命令

bin/kafka-server-start.sh -daemon config/server.properties

bin/kafka-server-stop.sh

1.3、集群遷移命令

• consumer.config <file>: 指定消費者的配置文件。該文件包含了消費源集群的相關配置，如連接信息、組ID等。
• producer.config <file>: 指定生產者的配置文件。該文件包含了生產者寫入目標集群的相關配置，如連接信息、壓縮方式等。
• whitelist <topic>: 指定需要鏡像的主題列表，可以使用逗號分隔多個主題名。
• blacklist <topic>: 指定需要排除鏡像的主題列表，可以使用逗號分隔多個主題名。與--whitelist互斥。
• num.streams <number>: 指定消費者線程數(shù)量，默認是1。
• message.handler <class>: 自定義消息處理類，可以在消息寫入目標集群前對其進行處理。
• message.handler.args <args>: 傳遞給自定義消息處理類的參數(shù)。
• abort.on.send.failure <true|false>: 是否在發(fā)送失敗時終止鏡像過程，默認為true。
• offset.commit.interval.ms <number>: 定期提交消費者偏移量的間隔時間，默認是60000ms（1分鐘）

#將指定topic按照消費者配置調用生產者發(fā)送到指定集群
bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist "topic1,topic2"

?

1.4、權限管理命令

1.4.1、權限參數(shù)介紹

• principal: 代表用戶或服務主體，例如 User:Alice。
• host: 允許或拒絕訪問的主機地址，* 表示任意主機。
• operation: 允許或拒絕的操作類型，例如 READ, WRITE, ALL。
• permissionType: 權限類型，可以是 ALLOW 或 DENY。
• resourceType: 資源類型，例如 TOPIC 或 GROUP。
• name: 資源名稱，例如 example-topic。
• patternType: 資源模式類型，LITERAL 表示字面模式。

1.4.2、增加權限命令

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \--remove --allow-principal User:user1 --operation Write --topic my-topic

1.4.3、移出權限命令

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \--remove --allow-principal User:user1 --operation Write --topic my-topic

1.4.4、查看所有topic權限命令

bin/kafka-acls.sh --list --bootstrap-server localhost:9092
Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=example-topic, patternType=LITERAL)`:(principal=User:Alice, host=*, operation=READ, permissionType=ALLOW)(principal=User:Bob, host=192.168.1.100, operation=WRITE, permissionType=ALLOW)(principal=User:Carol, host=*, operation=READ, permissionType=DENY)Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=another-topic, patternType=LITERAL)`:(principal=User:David, host=192.168.1.101, operation=READ, permissionType=ALLOW)Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=example-group, patternType=LITERAL)`:(principal=User:Admin, host=*, operation=ALL, permissionType=ALLOW)

1.4.5、查看某個topic權限命令

bin/kafka-acls.sh --authorizer-properties zookeeper.connect=localhost:2181 \--list --topic my-topic

2、生產者命令

2.1、創(chuàng)建topic命令

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

2.2、刪除topic命令

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

2.3、修改topic命令

2.3.1、修改配置命令

kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --config flush.messages=1

2.3.2、刪除配置命令

kafka-topics.sh --alter --bootstrap-server localhost:9092 --topic topicName --delete-config flush.messages

2.4、查詢topic命令

2.4.1、查詢topic列表命令

bin/kafka-topics.sh --list --bootstrap-server localhost:9092

2.4.2、查看某個topic詳情命令

bin/kafka-topics.sh --describe --topic my_topic --bootstrap-server localhost:9092

?2.4.2、查看某個topic配置命令

bin/kafka-configs.sh --bootstrap-server localhost:9092 --entity-type topics --entity-name my-topic --alter --add-config retention.ms=3600000Configs for topic 'my_topic' are:cleanup.policy=deletecompression.type=producerdelete.retention.ms=86400000file.delete.delay.ms=60000flush.messages=9223372036854775807flush.ms=9223372036854775807follower.replication.throttled.replicas=index.interval.bytes=4096leader.replication.throttled.replicas=max.message.bytes=1000012message.downconversion.enable=truemessage.format.version=2.7-IV0message.timestamp.difference.max.ms=9223372036854775807message.timestamp.type=CreateTimemin.cleanable.dirty.ratio=0.5min.compaction.lag.ms=0min.insync.replicas=1preallocate=falseretention.bytes=-1retention.ms=604800000segment.bytes=1073741824segment.index.bytes=10485760segment.jitter.ms=0segment.ms=604800000unclean.leader.election.enable=false

2.5、發(fā)送消息命令

#默認發(fā)送消息
bin/kafka-console-producer.sh --topic my-topic --bootstrap-server localhost:9092#指定分區(qū)發(fā)送命令
kafka-console-producer.sh --topic my-topic 2 --broker-list localhost:9092

3、消費者命令

3.1、消費者組命令

3.1.1、查看消費者組列表命令

#查看消費組列表命令
bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092group1
group2
group3
my_consumer_group
another_consumer_group

3.1.2、查看消費者組成員命令

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group test_group --membersCONSUMER-ID                                     HOST            CLIENT-ID       #PARTITIONS     
consumer-1-2d7d59c3-64c9-4526-905b-4e8920db3e84 /192.168.1.36   consumer-1      2

3.1.3、查看某個消費組詳情命令

bin/kafka-consumer-groups.sh --describe --group my_group --bootstrap-server localhost:9092GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID         HOST            CLIENT-ID
my_group        my_topic        0          12345           12350           5               consumer-1-1        /192.168.1.1    consumer-1
my_group        my_topic        1          23456           23460           4               consumer-1-2        /192.168.1.1    consumer-1
my_group        my_topic        2          34567           34567           0               consumer-1-3        /192.168.1.1    consumer-1

3.1.4、查看某個消費者組狀態(tài)命令

kafka-consumer-groups.sh --bootstrap-server localhost:9092 --describe --group my-group --stateCOORDINATOR (ID)          ASSIGNMENT-STRATEGY       STATE                #MEMBERS
192.168.1.36:9092 (1)     range                     Stable               1

3.3、消費消息命令

#從頭消費命令
bin/kafka-console-consumer.sh --topic my-topic --bootstrap-server localhost:9092 --from-beginning#指定分區(qū)消費命令
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --partition 1#指定消費者組id消費命令
kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning --consumer-property group.id=test

4、性能測試命令

4.1、生產者測試命令

bin/kafka-producer-perf-test.sh --topic perf-test --num-records 1000 --record-size 1024 --throughput -1 --producer-props bootstrap.servers=127.0.0.1:9092 compression.type=lz41000 records sent, 3424.657534 records/sec (3.34 MB/sec), 13.61 ms avg latency, 255.00 ms max latency, 13 ms 50th, 20 ms 95th, 255 ms 99th.-h, --help	顯示使用幫助并退出
--topic	指定生產的消息發(fā)往的 topic
--num-records	指定生產的消息總數(shù)
--payload-delimeter	如果通過 --payload-file 指定了從文件中獲取消息內容，那么這個參數(shù)的意義是指定文件的消息分隔符，默認值為 \n，即文件的每一行視為一條消息；如果未指定 --payload-file 則此參數(shù)不生效
--throughput	限制每秒發(fā)送的最大的消息數(shù)，設為 -1 表示不限制
--producer-props	直接指定 Producer 配置，格式為 NAME=VALUE，例如 bootstrap.server=127.0.0.1:9092，通過此種方式指定的配置優(yōu)先級高于 --producer.config
--producer-config	指定 Producer 的配置文件，格式參照官方的 config/producer.properties
--print-metrics	在測試結束后打印更詳盡的指標，默認為 false
--transactional-id	指定事務 ID，測試并發(fā)事務的性能時需要，只有在 --transaction-duration-ms > 0 時生效，默認值為 performance-producer-default-transactional-id
--transactional-duration-ms	指定事務持續(xù)的最長時間，超過這段時間后就會調用 commitTransaction 來提交事務，只有指定了 > 0 的值才會開啟事務，默認值為 0
--record-size	指定每條消息的大小，單位是字節(jié)，和 --payload-file 兩個中必須指定一個，但不能同時指定
--payload-file	指定消息的來源文件，只支持 UTF-8 編碼的文本文件，文件的消息分隔符通過 --payload-delimeter 指定，和 --record-size 兩個中必須指定一個，但不能同時指定

4.2、消費者測試命令

bin/kafka-consumer-perf-test.sh --bootstrap-server 127.0.0.1:9092 --topic perf_test --messages 1000000 --threads 8 --reporting-interval 1000 --show-detailedtime, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec, rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec
2021-03-25 15:57:59:426, 0, 657.2275, 657.2275, 673001, 673001.0000, 1616659078690, -1616659077690, 0.0000, 0.0000
...***************************輸入?yún)?shù)解釋***************************
--bootstrap-server	指定 broker 地址，必選，除非用 --broker-list 代替（不建議）
--topic	指定消費的 topic，必選
--version	輸出 Kafka 版本
--consumer.config	指定 Consumer 配置文件
--date-format	指定用于格式化 *.time 的規(guī)則，默認為 yyyy-MM-dd HH:mm:ss:SSS
--fetch-size	指定一次請求消費的大小，默認為 1048576 即 1 MB
--from-latest	如果 Consumer 沒有已經(jīng)建立的 offset，則指定從 log 中最新的位點開始消費，而不是從最早的位點開始消費
--group	指定 ConsumerGroup ID，默認為 perf-consumer-40924
--help	顯示使用幫助并退出
--hide-header	指定后不輸出 header 信息
--messages	指定消費的消息數(shù)量，必選
--num-fetch-threads	指定 fetcher 線程的數(shù)量
--print-metrics	指定打印 metrics 信息
--reporting-interval	指定打印進度信息的時間間隔，默認為 5000 即 5 秒
--show-detailed-stats	指定每隔一段時間（由 --reporting-interval 指定）輸出顯示詳細的狀態(tài)信息
--socket-buffer-size	指定 TCP 的 RECV 大小，默認為 2097152 即 2 MB
--threads	指定消費的線程數(shù)，默認為 10
--timeout	指定允許的最大超時時間，即每條消息返回的最大時間間隔，默認為 10000 即 10 秒***************************輸出參數(shù)解釋***************************
time：當前時間，格式由 --date-format 指定
threadId：線程 ID
data.consumed.in.MB：消費到的數(shù)據(jù)總大小，單位為 MB
MB.sec：消費 TPS，即每秒消費的消息大小
data.consumed.in.nMsg：消費到的總消息數(shù)
nMsg.sec：消費 TPS，即每秒消費的消息條數(shù)
rebalance.time.ms：消費者組重平衡的耗時，單位為 ms，0 表示沒有發(fā)生重平衡
fetch.time.ms：fetch 線程的總耗時，單位為 ms
fetch.MB.sec：fetch 線程每秒鐘獲取到的消息大小
fetch.nMsg.sec：fetch 線程每秒鐘獲取到的消息數(shù)量
#若沒有指定 --show-detailed，則輸出信息中的前兩項會有所不同，如下：
start.time, end.time, data.consumed.in.MB, MB.sec, ...
start.time：消費開始的時間，格式由 --date-format 指定
end.time：消費結束的時間，格式由 --date-format 指定

5、類執(zhí)行命令

通過kafka-run-class命令執(zhí)行指定類

#查看消費者消費進度
bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic my-topicmy-topic:0:12345
my-topic:1:23456
my-topic:2:34567