【云小課】應用平臺第6課 命令行管理Kafka的Topic，方便又實用

Kafka是一款非常熱門的消息中間件，它的消息生產與消費都圍繞消息主題（Topic）進行：生產者將消息發送給Topic，消息存儲在Topic的某個分區，消費者不斷從Topic拉取未讀消息進行處理。

華為云Kafka服務在控制臺提供Topic的部分管理功能，包括創建Topic、刪除Topic，以及Topic的老化時間、持久化與消息同步復制等功能設置。

在這里，小編向新手同學介紹Kafka提供的命令行工具，命令行工具相對于云服務控制臺，還提供了Topic的參數自定義、分區與副本數的增加等控制臺沒有提供的功能。

命令行工具準備

JDK環境配置

命令行工具依賴Java環境，需要先安裝JDK。以下介紹均以Linux環境為例。

JDK推薦使用Java Development Kit 1.8及以上版本。小編在華為云對象存儲（OBS）桶里存了一份JDK，供大家直接下載。Linux系統下可使用如下命令獲取：

wget https://407154.obs.cn-south-1.myhuaweicloud.com/jdk-8u231-linux-x64.tar.gz

JDK下載并解壓后可直接使用，但建議配置環境變量。小編的配置方式如下：

步驟 1 在環境變量參數中增加JAVA信息：vim ~/.bash_profile

補充如下內容：

export?JAVA_HOME=/root/jdk1.8.0_231??? ????export?PATH=$JAVA_HOME/bin:$PATH

步驟 2 使環境信息生效：source ~/.bash_profile

環境變量配置完成后，使用java -version驗證環境變量是否生效，版本號應與自定義的JDK一致。

下載Kafka命令行工具

Kafka安裝包自帶了命令行工具，推薦使用2.3.0以上版本，因為低版本的Kafka，Topic管理工具僅支持連接Kafka的Zookeeper組件，不支持直連Broker節點。

客戶端下載并解壓后即可使用。

wget https://archive.apache.org/dist/kafka/2.6.0/kafka_2.13-2.6.0.tgz

Kafka提供了服務管理、Topic管理、消息管理、消息遷移等30多個小工具，可以通過man手冊或者使用 --help選項了解工具的具體使用方法。以下我們重點介紹Topic管理相關的工具kafka-topics.sh（Windows下對應kafka-topics.bat）。

Kafka實例準備

命令行工具客戶端與環境準備了，還差Kafka服務：）如果您是華為云Kafka新用戶，華為云Kafka在華南-廣州區域為您提供了Kafka免費試用版本，省去環境搭建的繁瑣。

如下圖所示，登錄華為云控制臺后，找到分布式消息服務Kafka，右上角有一個“免費領取Kafka實例”的快捷入口。注意區域切換到“華南-廣州”。

為了方便本地客戶端訪問，在試用版的Kafka實例創建完成后，我為其開啟了公網訪問。在實例基本信息中，可查看訪問地址。

Topic管理

到現在為止，Kafka實例、命令行工具都已搭建好。以下介紹如何使用命令行工具kafka-topic.sh管理Kafka的Topic。kafka-topics.sh支持創建、查詢、刪除Topic，以及Topic的各類參數管理。

1.創建Topic

kafka-topics.sh --create --topic {topic_name} --bootstrap-server {broker_ip}:{port} --partitions {partition_num} --replication-factor {replication_num} --config conf_name1=conf_value1 --config conf_name2=conf_value2

create：創建Topic

bootstrap-server：Kafka的broker節點，Kafka集群有多broker節點，這里填1個即可。

partitions：Topic分區數量

replication-factor： 副本數，副本數量不能超過broker節點數量

config：Topic級別參數配置，以“名稱=值”的形式定義每一個參數

從華為云控制臺創建，部分Topic的配置參數不能自定義，如果您需要自定義這些配置參數，則可以嘗試通過命令行工具創建。

以下列出部分常用的配置參數：

cleanup.policy

消息強制持久化（fsync）的分區最大消息數量，比如此選項設置為10，則內存中某個分區達到10條消息則強制持久化到磁盤。

此參數設置需謹慎，過大則每次持久化的時間較長，且遇到故障后可能丟失的消息更多，過小則持久化頻繁則容易導致整體的客戶端請求有一定延遲。

2.查詢Topic列表

kafka-topics.sh --list --bootstrap-server {broker_ip}:{port}

list：查詢Topic列表，返回所有topic名稱

3.查看Topic的詳細信息

kafka-topics.sh --describe --topic {topic_name} --bootstrap-server {broker_ip}:{port}

describe：查看Topic詳情。

返回信息包括Topic名稱、分區數、副本數、Topic級別的配置信息，以及每一個分區的Leader副本所在broker節點等信息。

4.為Topic增加分區數、副本數等。

Kafka服務的控制臺界面目前沒有開放Topic的分區和副本數的修改功能，那么您可以通過命令行工具自助完成。

kafka-topics.sh --alter --partitions {partition_num} --bootstrap-server {broker_ip}:{port}? --topic {topic_name}

alter：對Topic進行配置變更。

變更命令需同時帶上Topic的某個參數，如分區數或者副本數。注意，修改分區數或副本數時，只能增加，不能減少。

5.刪除Topic

如果確定不再使用某個主題，最好的方式是刪除，這樣可以釋放一些資源，比如磁盤空間、文件句柄等。

kafka-topics.sh --delete --topic {topic_name} --bootstrap-server {broker_ip}:{port}

delete：刪除Topic。

注意，刪除主題操作不可逆，一旦刪除，與該主題相關的所有消息數據會被全部刪除。

6.修改Topic的配置

命令行工具使用broker節點地址修改Topic的配置時，存在bug，需要連接zookeeper地址才可以完成。由于華為云Kafka考慮穩定性等原因，沒有開放zookeeper，因此，小編建議您通過登錄Kafka-manager修改Topic的配置。將在下一期云小課中進行介紹。

云小課 分布式消息服務 Kafka

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。

版權聲明：本文內容由網絡用戶投稿，版權歸原作者所有，本站不擁有其著作權，亦不承擔相應法律責任。如果您發現本站中有涉嫌抄襲或描述失實的內容，請聯系我們jiasou666@gmail.com 處理，核實后本網站將在24小時內刪除侵權內容。