前言

通過前面 7 篇文章的介紹，小伙伴們應(yīng)該對(duì) Kafka 運(yùn)行工作原理有一個(gè)相對(duì)比較清晰的認(rèn)識(shí)了。

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

Kafka 安裝及簡單命令使用

Kafka中消息如何被存儲(chǔ)到Broker上？

Kafka消息發(fā)送時(shí)，網(wǎng)絡(luò)“偷偷”幫忙做的那點(diǎn)事兒

一文讀懂Kafka消費(fèi)者背后的那點(diǎn)"貓膩"

Kafka消息在服務(wù)端存儲(chǔ)與讀取

Kafka集群內(nèi)部工作原理的那些事

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

文章概覽

常用腳本及命令總結(jié)。

常用配置及說明。

常用命令總結(jié)

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

創(chuàng)建名字為 "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 的詳細(xì)信息，包括 Partition 個(gè)數(shù)，副本數(shù)，ISR 信息

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

修改指定 Topic 的 Partition 個(gè)數(shù)。注意：該命令只能增加 Partition 的個(gè)數(shù)，不能減少 Partition 個(gè)數(shù)，當(dāng)修改的 Partition 個(gè)數(shù)小于當(dāng)前的個(gè)數(shù)，會(huì)報(bào)如下錯(cuò)誤：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。注意如果直接執(zhí)行該命令，而沒有設(shè)置 delete.topic.enable 屬性為 true 時(shí)，是不會(huì)立即執(zhí)行刪除操作的，而是僅僅將指定的 Topic 標(biāo)記為刪除狀態(tài)，之后會(huì)啟動(dòng)后臺(tái)刪除線程來刪除。

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

二. kafka-consumer-groups.sh 腳本常用命令，主要用于操作消費(fèi)組相關(guān)的。

查看消費(fèi)端的所有消費(fèi)組信息(ZK 維護(hù)消費(fèi)信息)

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

查看消費(fèi)端的所有消費(fèi)組信息(Kafka 維護(hù)消費(fèi)信息)

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

查看指定 group 的詳細(xì)消費(fèi)情況，包括當(dāng)前消費(fèi)的進(jìn)度，消息積壓量等等信息(ZK 維護(hù)消息信息)

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

查看指定 group 的詳細(xì)消費(fèi)情況，包括當(dāng)前消費(fèi)的進(jìn)度，消息積壓量等等信息(Kafka 維護(hù)消費(fèi)信息)

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

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

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

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

查看指定 group 的 OffSet 信息，包括消費(fèi)者所在的 IP 信息等等

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

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

對(duì)指定 entry 添加配置，其中 entry 可以是 topics、clients；如下，給指定的 topic 添加兩個(gè)屬性，分別為 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'

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

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

查看指定 topic 的配置項(xiàng)。

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

五. kafka-reassign-partitions.sh 腳本相關(guān)常用命令，主要操作 Partition 的負(fù)載情況。

生成 partition 的移動(dòng)計(jì)劃，--broker-list 參數(shù)指定要挪動(dòng)的 Broker 的范圍，--topics-to-move-json-file 參數(shù)指定 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

執(zhí)行 Partition 的移動(dòng)計(jì)劃，--reassignment-json-file 參數(shù)指定 RePartition 后 Assigned Replica 的分布情況。

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

檢查當(dāng)前 rePartition 的進(jìn)度情況。

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

常用配置及說明

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

• Broker Config：kafka 服務(wù)端的配置
• Producer Config：生產(chǎn)端的配置
• Consumer Config：消費(fèi)端的配置
• Kafka Connect Config：kafka 連接相關(guān)的配置

Broker Config

zookeeper.connect

連接 zookeeper 集群的地址，用于將 kafka 集群相關(guān)的元數(shù)據(jù)信息存儲(chǔ)到指定的 zookeeper 集群中

advertised.port

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

auto.create.topics.enable

自動(dòng)創(chuàng)建 topic，默認(rèn)為 true。其作用是當(dāng)向一個(gè)還沒有創(chuàng)建的 topic 發(fā)送消息時(shí)，此時(shí)會(huì)自動(dòng)創(chuàng)建一個(gè) topic，并同時(shí)設(shè)置 -num.partition 1 (partition 個(gè)數(shù)) 和 default.replication.factor (副本個(gè)數(shù)，默認(rèn)為 1) 參數(shù)。

一般該參數(shù)需要手動(dòng)關(guān)閉，因?yàn)樽詣?dòng)創(chuàng)建會(huì)影響 topic 的管理，我們可以通過 kafka-topic.sh 腳本手動(dòng)創(chuàng)建 topic，通常也是建議使用這種方式創(chuàng)建 topic。在 0.10.x 之后提供了 kafka-admin 包，可以使用其來創(chuàng)建 topic。

auto.leader.rebalance.enable

自動(dòng) rebalance，默認(rèn)為 true。其作用是通過后臺(tái)線程定期掃描檢查，在一定條件下觸發(fā)重新 leader 選舉；在生產(chǎn)環(huán)境中，不建議開啟，因?yàn)樘鎿Q leader 在性能上沒有什么提升。

background.threads

后臺(tái)線程數(shù)，默認(rèn)為 10。用于后臺(tái)操作的線程，可以不用改動(dòng)。

broker.id

Broker 的唯一標(biāo)識(shí)，用于區(qū)分不同的 Broker。kafka 的檢查就是基于此 id 是否在 zookeeper 中/brokers/ids 目錄下是否有相應(yīng)的 id 目錄來判定 Broker 是否健康。

compression.type

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

delete.topic.enable

啟用刪除 topic。如果關(guān)閉，則無法使用 admin 工具進(jìn)行 topic 的刪除操作。

leader.imbalance.check.interval.seconds

partition 檢查重新 rebalance 的周期時(shí)間

leader.imbalance.per.broker.percentage

標(biāo)識(shí)每個(gè) Broker 失去平衡的比率，如果超過改比率，則執(zhí)行重新選舉 Broker 的 leader

log.dir / log.dirs

保存 kafka 日志數(shù)據(jù)的位置。如果 log.dirs 沒有設(shè)置，則使用 log.dir 指定的目錄進(jìn)行日志數(shù)據(jù)存儲(chǔ)。

log.flush.interval.messages

partition 分區(qū)的數(shù)據(jù)量達(dá)到指定大小時(shí)，對(duì)數(shù)據(jù)進(jìn)行一次刷盤操作。比如設(shè)置了 1024k 大小，當(dāng) partition 積累的數(shù)據(jù)量達(dá)到這個(gè)數(shù)值時(shí)則將數(shù)據(jù)寫入到磁盤上。

log.flush.interval.ms

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

log.retention.bytes

表示 topic 的容量達(dá)到指定大小時(shí)，則對(duì)其數(shù)據(jù)進(jìn)行清除操作，默認(rèn)為-1，永遠(yuǎn)不刪除。

log.retention.hours

標(biāo)示 topic 的數(shù)據(jù)最長保留多久，單位是小時(shí)

log.retention.minutes

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

log.retention.ms

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

log.roll.hours

新的 segment 創(chuàng)建周期，單位小時(shí)。kafka 數(shù)據(jù)是以 segment 存儲(chǔ)的，當(dāng)周期時(shí)間到達(dá)時(shí)，就創(chuàng)建一個(gè)新的 segment 來存儲(chǔ)數(shù)據(jù)。

log.segment.bytessegment 的大小。當(dāng) segment 大小達(dá)到指定值時(shí)，就新創(chuàng)建一個(gè) segment。

message.max.bytes

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

min.insync.replicas

最小副本同步個(gè)數(shù)。當(dāng) producer 設(shè)置了 request.required.acks 為-1 時(shí)，則 topic 的副本數(shù)要同步至該參數(shù)指定的個(gè)數(shù)，如果達(dá)不到，則 producer 端會(huì)產(chǎn)生異常。

num.io.threads

指定 io 操作的線程數(shù)

num.network.threads

執(zhí)行網(wǎng)絡(luò)操作的線程數(shù)

num.recovery.threads.per.data.dir

每個(gè)數(shù)據(jù)目錄用于恢復(fù)數(shù)據(jù)的線程數(shù)

num.replica.fetchers

從 leader 備份數(shù)據(jù)的線程數(shù)

offset.metadata.max.bytes

允許消費(fèi)者端保存 offset 的最大個(gè)數(shù)

offsets.commit.timeout.ms

offset 提交的延遲時(shí)間

offsets.topic.replication.factor

topic 的 offset 的備份數(shù)量。該參數(shù)建議設(shè)置更高保證系統(tǒng)更高的可用性

port

端口號(hào)，Broker 對(duì)外提供訪問的端口號(hào)。

request.timeout.ms

Broker 接收到請(qǐng)求后的最長等待時(shí)間，如果超過設(shè)定時(shí)間，則會(huì)給客戶端發(fā)送錯(cuò)誤信息

zookeeper.connection.timeout.ms

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

connections.max.idle.ms

空連接的超時(shí)時(shí)間。即空閑連接超過該時(shí)間時(shí)則自動(dòng)銷毀連接。

Producer Config

bootstrap.servers

服務(wù)端列表。即接收生產(chǎn)消息的服務(wù)端列表

key.serializer

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

value.serializer

消息內(nèi)容的序列化方式。指定 value 的序列化類型

acks

消息寫入 Partition 的個(gè)數(shù)。通常可以設(shè)置為 0，1，all；當(dāng)設(shè)置為 0 時(shí)，只需要將消息發(fā)送完成后就完成消息發(fā)送功能；當(dāng)設(shè)置為 1 時(shí)，即 Leader Partition 接收到數(shù)據(jù)并完成落盤；當(dāng)設(shè)置為 all 時(shí)，即主從 Partition 都接收到數(shù)據(jù)并落盤。

buffer.memory

客戶端緩存大小。即 Producer 生產(chǎn)的數(shù)據(jù)量緩存達(dá)到指定值時(shí)，將緩存數(shù)據(jù)一次發(fā)送的 Broker 上。

compression.type

壓縮類型。指定消息發(fā)送前的壓縮類型，通常有 none, gzip, snappy, or, lz4 四種。不指定時(shí)消息默認(rèn)不壓縮。

retries

消息發(fā)送失敗時(shí)重試次數(shù)。當(dāng)該值設(shè)置大于 0 時(shí)，消息因?yàn)榫W(wǎng)絡(luò)異常等因素導(dǎo)致消息發(fā)送失敗進(jìn)行重新發(fā)送的次數(shù)。

Consumer Config

bootstrap.servers

服務(wù)端列表。即消費(fèi)端從指定的服務(wù)端列表中拉取消息進(jìn)行消費(fèi)。

key.deserializer

消息鍵的反序列化方式。指定 key 的反序列化類型，與序列化時(shí)指定的類型相對(duì)應(yīng)。

value.deserializer

消息內(nèi)容的反序列化方式。指定 value 的反序列化類型，與序列化時(shí)指定的類型相對(duì)應(yīng)。

fetch.min.bytes

抓取消息的最小內(nèi)容。指定每次向服務(wù)端拉取的最小消息量。

group.id

消費(fèi)組中每個(gè)消費(fèi)者的唯一表示。

heartbeat.interval.ms

心跳檢查周期。即在周期性的向 group coordinator 發(fā)送心跳，當(dāng)服務(wù)端發(fā)生 rebalance 時(shí)，會(huì)將消息發(fā)送給各個(gè)消費(fèi)者。該參數(shù)值需要小于 session.timeout.ms，通常為后者的 1/3。

max.partition.fetch.bytes

Partition 每次返回的最大數(shù)據(jù)量大小。

session.timeout.ms

consumer 失效的時(shí)間。即 consumer 在指定的時(shí)間后，還沒有響應(yīng)則認(rèn)為該 consumer 已經(jīng)發(fā)生故障了。

auto.offset.reset

當(dāng) kafka 中沒有初始偏移量或服務(wù)器上不存在偏移量時(shí)，指定從哪個(gè)位置開始消息消息。earliest：指定從頭開始；latest：從最新的數(shù)據(jù)開始消費(fèi)。

Kafka Connect Config

group.id

消費(fèi)者在消費(fèi)組中的唯一標(biāo)識(shí)

internal.key.converter

內(nèi)部 key 的轉(zhuǎn)換類型。

internal.value.converter

內(nèi)部 value 的轉(zhuǎn)換類型。

key.converter

服務(wù)端接收到 key 時(shí)指定的轉(zhuǎn)換類型。

value.converter

服務(wù)端接收到 value 時(shí)指定的轉(zhuǎn)換類型。

bootstrap.servers

服務(wù)端列表。

heartbeat.interval.ms

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

session.timeout.ms

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

總結(jié)

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

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

---

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。