運行環境：mac os

1.?啟動zookeeper

./bin/zookeeper-server-start /usr/local/etc/kafka/zookeeper.properties

2.?啟動kafka服務

./bin/kafka-server-start /usr/local/etc/kafka/server.properties

3.?查看topic列表

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

4. 創建topic

./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 --partitions 3 --topic hezhixiong--create：指定創建topic動作 --zookeeper：指定kafka連接zk的地址 --replication-factor：指定每個分區的復制因子個數，默認1個 --partitions：指定當前創建的kafka分區數量，默認為1個 --topic：指定新建topic的名稱

5. 查看topic的描述信息

./bin/kafka-topics --describe --zookeeper localhost:2181 --topic hezhixiong

6. 修改topic信息

./bin/kafka-topics --zookeeper localhost:2181 --alter --topic hezhixiong --partitions 4 <!-- Kafka分區數量只允許增加，不允許減少 -->

7. 刪除topic

./bin/kafka-topics --zookeeper localhost:2181 --delete --topic hezhixiong Topic hezhixiong is marked for deletion. Note: This will have no impact if delete.topic.enable is not set to true.默認情況下Kafka的Topic是沒法直接刪除的，而是標記刪除而已。如果需要徹底刪除topic，有以下兩種方式： 1. 通過delete命令刪除后，手動將本地磁盤以及zk上的相關topic的信息刪除 2. 配置server.properties文件，給定參數delete.topic.enable=true，重啟kafka服務，此時執行delete命令表示允許進行Topic的刪除

8. 生產者

./bin/kafka-console-producer --broker-list localhost:9092 --topic hezhixiong

9. 消費者

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

?10. broker集群

　　首先為每個broker創建一個配置文件：

cd /usr/local/etc/kafka cp server.properties server_1.properties cp server.properties server_2.propertiesvi server_1.propertiesbroker.id=1 listeners=PLAINTEXT://:9093 log.dir=/tmp/kafka-logs/one vi server_2.propertiesbroker.id=2 listeners=PLAINTEXT://:9094 log.dir=/tmp/kafka-logs/two

　　broker.id是集群中每個節點的唯一且永久的名稱，我們修改端口和日志目錄是因為我們現在在同一臺機器上運行，我們要防止broker在同一端口上注冊和覆蓋對方的數據。

　　我們已經運行了zookeeper和剛才的一個kafka節點，所有我們只需要在啟動2個新的kafka節點。

./bin/kafka-server-start /usr/local/etc/kafka/server_1.properties ./bin/kafka-server-start /usr/local/etc/kafka/server_2.properties

　　現在，我們創建一個新topic，把備份設置為3

./bin/kafka-topics --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic hzx-new
./bin/kafka-topics --describe --zookeeper localhost:2181 --topic hzx-new 查看topic得到輸出如下：
Topic:hzx-new PartitionCount:1 ReplicationFactor:3 Configs:
Topic: hzx-new Partition: 0 Leader: 2 Replicas: 2,1,0 Isr: 2,1,0

從上面的輸出，我們可以得出結論，第一行給出所有分區的摘要，顯示主題名稱，分區數量和我們已經選擇的復制因子。在第二行中，每個節點將是分區的隨機選擇部分的領導者。

在我的例子中，我看到的第一個broker(with broker.id 2)是領導者。 然后Replicas:2，1，0 意味著所有代理復制主題。最后 Isr 是 in-sync 副本的集合。 那么，這是副本的子集，當前活著并被領導者趕上。

"leader"：該節點負責該分區的所有的讀和寫，每個節點的leader都是隨機選擇的。
"replicas"：備份的節點列表，無論該節點是否是leader或者目前是否還活著，只是顯示。
"isr"：“同步備份”的節點列表，也就是活著的節點并且正在同步leader。

　　啟動生產者和消費者，在生產者端發送消息，在消費者端能看到消息的消息

./bin/kafka-console-producer --broker-list localhost:9092 --topic hzx-new （生產者） ./bin/kafka-console-consumer --bootstrap-server localhost:9092 --topic hzx-new --from-beginning （消費者）

?　　體驗并測試kafka集群的容錯，目前Leader是broker2，所以結束調broker2，并查看topic：hzx-new 的信息，輸出如下

Topic:hzx-new PartitionCount:1 ReplicationFactor:3 Configs:Topic: hzx-new Partition: 0 Leader: 1 Replicas: 2,1,0 Isr: 1,0可以看出 broker2已經不再同步備份集合里了，備份節點之一的broker1成為了新的leader了。

?

參考資料：

http://orchome.com/6

轉載于:https://www.cnblogs.com/hezhixiong/p/10078994.html

總結

以上是生活随笔為你收集整理的kafka operation的全部內容，希望文章能夠幫你解決所遇到的問題。

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