前言

通過前面 7 篇文章的介紹，小伙伴們應該對 Kafka 運行工作原理有一個相對比較清晰的認識了。

Kafka是什么？一起來看看吧！

Kafka 安裝及簡單命令使用

Kafka中消息如何被存儲到Broker上？

Kafka消息發送時，網絡“偷偷”幫忙做的那點事兒

一文讀懂Kafka消費者背后的那點"貓膩"

Kafka消息在服務端存儲與讀取

Kafka集群內部工作原理的那些事

為了提高平時的工作效率，幫助我們快速定位一些線上問題，比如查看部分 Partition 堆積機器 IP 等操作，這篇文章總結了一些平時常用到的一些 Kafka 命令及常用配置，方便日后查閱(該文章中提到的相關配置會持續更新)。

文章概覽

常用腳本及命令總結。

常用配置及說明。

常用命令總結

一. kafka-topic.sh 腳本相關常用命令，主要操作 Topic。

創建名字為 "op_log" 的 Topic。

$ bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition 3 --topic op_log

查看指定 ZK?管理的 Topic 列表

$ bin/kafka-topics.sh --list --zookeeper localhost:2181

查看指定 Topic 的詳細信息，包括 Partition 個數，副本數，ISR 信息

$ bin/kafka-topics.sh --zookeeper localhost:2181 --describe op_log

修改指定 Topic 的 Partition 個數。注意：該命令只能增加 Partition 的個數，不能減少 Partition 個數，當修改的 Partition 個數小于當前的個數，會報如下錯誤：Error while executing topic command : The number of partitions for a topic can only be increased

$ bin/kafka-topics.sh --alter --zookeeper localhost:2181 --topic op_log1 --partition 4

刪除名字為 "op_log" 的 Topic。注意如果直接執行該命令，而沒有設置 delete.topic.enable 屬性為 true 時，是不會立即執行刪除操作的，而是僅僅將指定的 Topic 標記為刪除狀態，之后會啟動后臺刪除線程來刪除。

$ bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic op_log

二. kafka-consumer-groups.sh 腳本常用命令，主要用于操作消費組相關的。

查看消費端的所有消費組信息(ZK 維護消費信息)

$ bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --list

查看消費端的所有消費組信息(Kafka 維護消費信息)

$ bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --list

查看指定 group 的詳細消費情況，包括當前消費的進度，消息積壓量等等信息(ZK 維護消息信息)

$ bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --group console-consumer-1291 --describe

查看指定 group 的詳細消費情況，包括當前消費的進度，消息積壓量等等信息(Kafka 維護消費信息)

$ bin/kafka-consumer-groups.sh --new-consumer --bootstrap-server localhost:9092 --group mygroup --describe

三. kafka-consumer-offset-checker.sh 腳本常用命令，用于檢查 OffSet 相關信息。(注意：該腳本在 0.9 以后可以使用 kafka-consumer-groups.sh 腳本代替，官方已經標注為 deprecated 了)

查看指定 group 的 OffSet 信息、所有者等信息

$ bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --topic mytopic --group console-consumer-1291

查看指定 group 的 OffSet 信息，包括消費者所在的 IP 信息等等

$ bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --topic mytopic --group console-consumer-1291 --broker-info

四. kafka-configs.sh 腳本常用命令，該腳本主要用于增加/修改 Kafka 集群的配置信息。

對指定 entry 添加配置，其中 entry 可以是 topics、clients；如下，給指定的 topic 添加兩個屬性，分別為 max.message.byte、flush.message。

$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name mytopic --add-config 'max.message.bytes=50000000,flush.message=5'

對指定 entry 刪除配置，其中 entry 可以是 topics、clients；如下。對指定的 topic 刪除 flush.message 屬性。

$ bin/kafka-configs.sh --zookeeper localhost:2181 --alter --entity-type topics --entity-name mytopic --delete-config 'flush.message'

查看指定 topic 的配置項。

$ bin/kafka-configs.sh --zookeeper localhost:2181 --entity-type topics --entity-name mytopic --describe

五. kafka-reassign-partitions.sh 腳本相關常用命令，主要操作 Partition 的負載情況。

生成 partition 的移動計劃，--broker-list 參數指定要挪動的 Broker 的范圍，--topics-to-move-json-file 參數指定 Json 配置文件

$ bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --broker-list "0" --topics-to-move-json-file ~/json/op_log_topic-to-move.json --generate

執行 Partition 的移動計劃，--reassignment-json-file 參數指定 RePartition 后 Assigned Replica 的分布情況。

$ bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/json/op_log_reassignment.json --execute

檢查當前 rePartition 的進度情況。

$ bin/kafka-reassign-partitions.sh --zookeeper localhost:2181 --reassignment-json-file ~/json/op_log_reassignment.json --verify

常用配置及說明

kafka 常見重要配置說明，分為四部分

• Broker Config：kafka 服務端的配置
• Producer Config：生產端的配置
• Consumer Config：消費端的配置
• Kafka Connect Config：kafka 連接相關的配置

Broker Config

zookeeper.connect

連接 zookeeper 集群的地址，用于將 kafka 集群相關的元數據信息存儲到指定的 zookeeper 集群中

advertised.port

注冊到 zookeeper 中的地址端口信息，在 IaaS 環境中，默認注冊到 zookeeper 中的是內網地址，通過該配置指定外網訪問的地址及端口，advertised.host.name 和 advertised.port 作用和 advertised.port 差不多，在 0.10.x 之后，直接配置 advertised.port 即可，前兩個參數被廢棄掉了。

auto.create.topics.enable

自動創建 topic，默認為 true。其作用是當向一個還沒有創建的 topic 發送消息時，此時會自動創建一個 topic，并同時設置 -num.partition 1 (partition 個數) 和 default.replication.factor (副本個數，默認為 1) 參數。

一般該參數需要手動關閉，因為自動創建會影響 topic 的管理，我們可以通過 kafka-topic.sh 腳本手動創建 topic，通常也是建議使用這種方式創建 topic。在 0.10.x 之后提供了 kafka-admin 包，可以使用其來創建 topic。

auto.leader.rebalance.enable

自動 rebalance，默認為 true。其作用是通過后臺線程定期掃描檢查，在一定條件下觸發重新 leader 選舉；在生產環境中，不建議開啟，因為替換 leader 在性能上沒有什么提升。

background.threads

后臺線程數，默認為 10。用于后臺操作的線程，可以不用改動。

broker.id

Broker 的唯一標識，用于區分不同的 Broker。kafka 的檢查就是基于此 id 是否在 zookeeper 中/brokers/ids 目錄下是否有相應的 id 目錄來判定 Broker 是否健康。

compression.type

壓縮類型。此配置可以接受的壓縮類型有 gzip、snappy、lz4。另外可以不設置，即保持和生產端相同的壓縮格式。

delete.topic.enable

啟用刪除 topic。如果關閉，則無法使用 admin 工具進行 topic 的刪除操作。

leader.imbalance.check.interval.seconds

partition 檢查重新 rebalance 的周期時間

leader.imbalance.per.broker.percentage

標識每個 Broker 失去平衡的比率，如果超過改比率，則執行重新選舉 Broker 的 leader

log.dir / log.dirs

保存 kafka 日志數據的位置。如果 log.dirs 沒有設置，則使用 log.dir 指定的目錄進行日志數據存儲。

log.flush.interval.messages

partition 分區的數據量達到指定大小時，對數據進行一次刷盤操作。比如設置了 1024k 大小，當 partition 積累的數據量達到這個數值時則將數據寫入到磁盤上。

log.flush.interval.ms

數據寫入磁盤時間間隔，即內存中的數據保留多久就持久化一次，如果沒有設置，則使用 log.flush.scheduler.interval.ms 參數指定的值。

log.retention.bytes

表示 topic 的容量達到指定大小時，則對其數據進行清除操作，默認為-1，永遠不刪除。

log.retention.hours

標示 topic 的數據最長保留多久，單位是小時

log.retention.minutes

表示 topic 的數據最長保留多久，單位是分鐘，如果沒有設置該參數，則使用 log.retention.hours 參數

log.retention.ms

表示 topic 的數據最長保留多久，單位是毫秒，如果沒有設置該參數，則使用 log.retention.minutes 參數

log.roll.hours

新的 segment 創建周期，單位小時。kafka 數據是以 segment 存儲的，當周期時間到達時，就創建一個新的 segment 來存儲數據。

log.segment.bytessegment 的大小。當 segment 大小達到指定值時，就新創建一個 segment。

message.max.bytes

topic 能夠接收的最大文件大小。需要注意的是 producer 和 consumer 端設置的大小需要一致。

min.insync.replicas

最小副本同步個數。當 producer 設置了 request.required.acks 為-1 時，則 topic 的副本數要同步至該參數指定的個數，如果達不到，則 producer 端會產生異常。

num.io.threads

指定 io 操作的線程數

num.network.threads

執行網絡操作的線程數

num.recovery.threads.per.data.dir

每個數據目錄用于恢復數據的線程數

num.replica.fetchers

從 leader 備份數據的線程數

offset.metadata.max.bytes

允許消費者端保存 offset 的最大個數

offsets.commit.timeout.ms

offset 提交的延遲時間

offsets.topic.replication.factor

topic 的 offset 的備份數量。該參數建議設置更高保證系統更高的可用性

port

端口號，Broker 對外提供訪問的端口號。

request.timeout.ms

Broker 接收到請求后的最長等待時間，如果超過設定時間，則會給客戶端發送錯誤信息

zookeeper.connection.timeout.ms

客戶端和 zookeeper 建立連接的超時時間，如果沒有設置該參數，則使用 zookeeper.session.timeout.ms 值

connections.max.idle.ms

空連接的超時時間。即空閑連接超過該時間時則自動銷毀連接。

Producer Config

bootstrap.servers

服務端列表。即接收生產消息的服務端列表

key.serializer

消息鍵的序列化方式。指定 key 的序列化類型

value.serializer

消息內容的序列化方式。指定 value 的序列化類型

acks

消息寫入 Partition 的個數。通常可以設置為 0，1，all；當設置為 0 時，只需要將消息發送完成后就完成消息發送功能；當設置為 1 時，即 Leader Partition 接收到數據并完成落盤；當設置為 all 時，即主從 Partition 都接收到數據并落盤。

buffer.memory

客戶端緩存大小。即 Producer 生產的數據量緩存達到指定值時，將緩存數據一次發送的 Broker 上。

compression.type

壓縮類型。指定消息發送前的壓縮類型，通常有 none, gzip, snappy, or, lz4 四種。不指定時消息默認不壓縮。

retries

消息發送失敗時重試次數。當該值設置大于 0 時，消息因為網絡異常等因素導致消息發送失敗進行重新發送的次數。

Consumer Config

bootstrap.servers

服務端列表。即消費端從指定的服務端列表中拉取消息進行消費。

key.deserializer

消息鍵的反序列化方式。指定 key 的反序列化類型，與序列化時指定的類型相對應。

value.deserializer

消息內容的反序列化方式。指定 value 的反序列化類型，與序列化時指定的類型相對應。

fetch.min.bytes

抓取消息的最小內容。指定每次向服務端拉取的最小消息量。

group.id

消費組中每個消費者的唯一表示。

heartbeat.interval.ms

心跳檢查周期。即在周期性的向 group coordinator 發送心跳，當服務端發生 rebalance 時，會將消息發送給各個消費者。該參數值需要小于 session.timeout.ms，通常為后者的 1/3。

max.partition.fetch.bytes

Partition 每次返回的最大數據量大小。

session.timeout.ms

consumer 失效的時間。即 consumer 在指定的時間后，還沒有響應則認為該 consumer 已經發生故障了。

auto.offset.reset

當 kafka 中沒有初始偏移量或服務器上不存在偏移量時，指定從哪個位置開始消息消息。earliest：指定從頭開始；latest：從最新的數據開始消費。

Kafka Connect Config

group.id

消費者在消費組中的唯一標識

internal.key.converter

內部 key 的轉換類型。

internal.value.converter

內部 value 的轉換類型。

key.converter

服務端接收到 key 時指定的轉換類型。

value.converter

服務端接收到 value 時指定的轉換類型。

bootstrap.servers

服務端列表。

heartbeat.interval.ms

心跳檢測，與 consumer 中指定的配置含義相同。

session.timeout.ms

session 有效時間，與 consumer 中指定的配置含義相同。

總結

本文總結了平時經常用到的一些 Kafka 配置及命令說明，方便隨時查看；喜歡的朋友可以收藏以備不時之需。

下篇文章我們來分析一些經常在面試中碰到的問題及相應的解決辦法，敬請期待。

---

總結

以上是生活随笔為你收集整理的kafka创建topic_一网打尽Kafka常用命令、脚本及配置，宜收藏！的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。