Kafka 單節點單Kafka Broker集群

$ tar zxvf kafka_2.12-0.10.2.1.tgz

$ cd kafka_2.12-0.10.2.1

$ bin/zookeeper-server-start.sh config/zookeeper.properties

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

首先，建立乙個單分割槽單副本的 topic: mytopic

$ bin/kafka -topics.sh --create-- zookeeper localhost:2181-- replication -factor1- -partitions1- -topic mytopic created topic "mytopic"

.

然後，可以通過執行 list 命令來檢視已經存在的 topic，比如：

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

mytopic

也可以使用 describe 命令來檢視。由於我們現在是單分割槽單副本的case，所以 leader 和 isr （複製節點集合）都只在broker 0上。

bin/kafka -topics.sh --describe-- zookeeper localhost:2181-- topic mytopic topic:mytopic partitioncount:1 replicationfactor:1 configs: topic: mytopic partition: 0leader: 0replicas: 0isr:

0

現在，我們通過 kafka 自帶命令列客戶端向這個 topic 傳送訊息。

$ bin/kafka-console -producer .sh --broker-list localhost:9092 --topic mytopic aaabbb

ccc

另開乙個終端，然後使用 kafka 自帶的命令列工具來消費這些訊息。

$ bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic mytopic --from-beginning ...aaa

bbbccc

此時可以在 producer 命令列視窗繼續輸入訊息，然後觀察 consumer 終端視窗，可以看到它們被消費列印出來。

下面使用 kafka 的 connect 演示從乙個變化的檔案中讀取增量資料然後輸出到另外乙個檔案中。

首先執行下面的指令碼，此指令碼會每隔一秒會向 test.txt 檔案中追加乙個數字。

$ for i in ;do
echo
$i >> test.txt; sleep 1; done

然後執行下面的指令碼，此時生產者從 test.txt 檔案中讀取檔案內容並作為訊息傳送到topic中，然後消費者從topic中消費訊息並輸出到 test.sink.txt 檔案中。命令列使用的配置檔案中定義了輸入輸出的檔名和使用的topic名。

$ bin/connect-standalone .sh config/connect-standalone .properties config/connect-file -source .properties config/connect-file -sink

.properties

執行後，另開乙個終端來觀察 test.sink.txt 檔案中的內容，可以看到檔案中的內容會不停的增加。

要刪除 topic，可以使用下面的命令

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

mytopic

但是執行命令後，topic並沒有被刪除，使用「bin/kafka-topics.sh –list –zookeeper localhost:2181」仍然可以查到。此時我們需要修改config/server.properties檔案中的「delete.topic.enable=true」來開啟這個功能。此時我們再執行上面的 –delete 操作，即可刪除topic了。

Kafka 單節點單Kafka Broker集群

kafka單節點測試

Kafka 單節點多Kafka Broker集群

kafka單節點broker安裝

Kafka 單節點單Kafka Broker集群

kafka單節點測試

Kafka 單節點多Kafka Broker集群

kafka單節點broker安裝

相關推薦