Kafka快速開始（安裝部署）

 

第1步：下載代碼

下載 2.0.0版本並解壓縮它。

 

1 2	> tar -xzf kafka_2.11-2.0.0.tgz > cd kafka_2.11-2.0.0

 

 

第2步：啓動服務器

Kafka使用ZooKeeper，所以若是您尚未ZooKeeper服務器，則須要先啓動它。您可使用與kafka一塊兒打包的便捷腳原本獲取的單節點ZooKeeper實例。

 

1 2 3

> bin/zookeeper-server-start.sh config/zookeeper.properties [2013-04-22 15:01:37,495] INFO Reading configuration from: config/zookeeper.properties (org.apache.zookeeper.server.quorum.QuorumPeerConfig) ...

 

如今啓動Kafka服務器：

 

1 2 3 4

> bin/kafka-server-start.sh config/server.properties [2013-04-22 15:01:47,028] INFO Verifying properties (kafka.utils.VerifiableProperties) [2013-04-22 15:01:47,051] INFO Property socket.send.buffer.bytes is overridden to 1048576 (kafka.utils.VerifiableProperties) ...

 

第3步：建立主題

讓咱們建立一個名爲「test」的主題，它只包含一個分區，只有一個副本：

 

1	> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

 

若是咱們運行list topic命令，咱們如今能夠看到該主題：

 

1 2	> bin/kafka-topics.sh --list --zookeeper localhost:2181 test

 

或者，您也能夠將代理配置爲在發佈不存在的主題時自動建立主題，而不是手動建立主題。

第4步：發送一些消息

Kafka附帶一個命令行客戶端，它將從文件或標準輸入中獲取輸入，並將其做爲消息發送到Kafka集羣。默認狀況下，每行將做爲單獨的消息發送。

運行producer，而後在控制檯中鍵入一些消息以發送到服務器。

 

1 2 3

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test This is a message This is another message

 

第5步：啓動消費者

Kafka還有一個命令行consumer，它會將消息轉儲到標準輸出。

 

1 2 3

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning This is a message This is another message

 

若是您在不一樣的終端中運行上述每一個命令，那麼您如今應該可以在生產者終端中鍵入消息並看到它們出如今消費者終端中。

全部命令行工具都有其餘選項; 運行不帶參數的命令將顯示更詳細地記錄它們的使用信息。

步驟6：設置多代理羣集

到目前爲止，咱們一直只運行一個單broker，但這並很差玩。對於Kafka，單個broker只是一個大小爲1的集羣，所以除了啓動更多代理實例以外沒有太多變化。可是爲了感覺它，讓咱們將咱們的集羣擴展到三個節點（仍然在咱們的本地機器上）。

首先，咱們爲每一個代理程序建立一個配置文件（在Windows上使用copy命令代替）：

 

1 2	> cp config/server.properties config/server-1.properties > cp config/server.properties config/server-2.properties

 

如今編輯這些新文件並設置如下屬性：

 

1 2 3 4 5 6 7 8 9

config/server-1.properties:     broker.id=1     listeners= PLAINTEXT://:9093     log.dirs=/tmp/kafka-logs-1   config/server-2.properties:     broker.id=2     listeners= PLAINTEXT://:9094     log.dirs=/tmp/kafka-logs-2

 

該broker.id屬性是羣集中每一個節點的惟一且永久的名稱。咱們必須覆蓋端口和日誌目錄，由於咱們在同一臺機器上運行這些，而且咱們但願讓全部代理嘗試在同一端口上註冊或覆蓋彼此的數據。

咱們已經啓動了Zookeeper並啓動了咱們的單個節點，所以咱們只須要啓動兩個新節點：

 

1 2 3 4

> bin/kafka-server-start.sh config/server-1.properties & ... > bin/kafka-server-start.sh config/server-2.properties & ...

 

如今建立一個複製因子爲3的新主題：

 

1	> bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

 

好的，但如今咱們有一個集羣，咱們怎麼知道哪一個經紀人正在作什麼？要查看運行「describe topics」命令：

 

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:     Topic: my-replicated-topic  Partition: 0    Leader: 1   Replicas: 1,2,0 Isr: 1,2,0

 

這是輸出的解釋。第一行給出了全部分區的摘要，每一個附加行提供有關一個分區的信息。因爲咱們只有一個分區用於此主題，所以只有一行。

• 「leader」是負責給定分區的全部讀取和寫入的節點。每一個節點將成爲隨機選擇的分區部分的領導者。
• 「replicas」是複製此分區日誌的節點列表，不管它們是否爲領導者，或者即便它們當前處於活動狀態。
• 「isr」是「同步」複製品的集合。這是副本列表的子集，該列表當前處於活躍狀態而且已經被領導者捕獲。

請注意，在個人示例中，節點1是該主題的惟一分區的領導者。

咱們能夠在咱們建立的原始主題上運行相同的命令，以查看它的位置：

 

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic test Topic:test  PartitionCount:1    ReplicationFactor:1 Configs:     Topic: test Partition: 0    Leader: 0   Replicas: 0 Isr: 0

 

因此毫無疑問 - 原始主題沒有副本，位於服務器0上，是咱們建立它時羣集中惟一的服務器。

讓咱們向咱們的新主題發佈一些消息：

 

1 2 3 4 5

> bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic ... my test message 1 my test message 2 ^C

 

如今讓咱們使用這些消息：

 

1 2 3 4 5

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C

 

如今讓咱們測試一下容錯性。broker1充當leader因此讓咱們kill掉它的進程：

 

1 2 3

> ps aux | grep server-1.properties 7564 ttys002    0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.8/Home/bin/java... > kill -9 7564

 

在Windows上使用：

 

1 2 3 4

> wmic process where "caption = 'java.exe' and commandline like '%server-1.properties%'" get processid ProcessId 6016 > taskkill /pid 6016 /f

 

leader已切換到其中一個從屬節點，節點1再也不位於同步副本集中：

 

1 2 3

> bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic Topic:my-replicated-topic   PartitionCount:1    ReplicationFactor:3 Configs:     Topic: my-replicated-topic  Partition: 0    Leader: 2   Replicas: 1,2,0 Isr: 2,0

 

但即便最初接受寫入的領導者已經失敗，這些消息仍可供消費：

 

1 2 3 4 5

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ^C

 

步驟7：使用Kafka Connect導入/導出數據

從控制檯寫入數據並將其寫回控制檯是一個方便的起點，但您可能但願使用其餘來源的數據或將數據從Kafka導出到其餘系統。對於許多系統，您可使用Kafka Connect導入或導出數據，而不是編寫自定義集成代碼。

Kafka Connect是Kafka附帶的工具，用於向Kafka導入和導出數據。它是一個可擴展的工具，運行 鏈接器，實現與外部系統交互的自定義邏輯。在本快速入門中，咱們將瞭解如何使用簡單的鏈接器運行Kafka Connect，這些鏈接器將數據從文件導入Kafka主題並將數據從Kafka主題導出到文件。

首先，咱們將首先建立一些種子數據進行測試：

 

1	> echo -e "foo\nbar" > test.txt

 

或者在Windows上：

 

1 2	> echo foo> test.txt > echo bar>> test.txt

 

接下來，咱們將啓動兩個以獨立模式運行的鏈接器，這意味着它們在單個本地專用進程中運行。咱們提供三個配置文件做爲參數。第一個始終是Kafka Connect流程的配置，包含常見配置，例如要鏈接的Kafka代理和數據的序列化格式。其他配置文件均指定要建立的鏈接器。這些文件包括惟一的鏈接器名稱，要實例化的鏈接器類以及鏈接器所需的任何其餘配置。

 

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

 

Kafka附帶的這些示例配置文件使用您以前啓動的默認本地羣集配置並建立兩個鏈接器：第一個是源鏈接器，它從輸入文件讀取行並生成每一個Kafka主題，第二個是宿鏈接器從Kafka主題讀取消息並將每一個消息生成爲輸出文件中的一行。

在啓動過程當中，您將看到許多日誌消息，包括一些指示正在實例化鏈接器的日誌消息。一旦Kafka Connect進程啓動，源鏈接器應該開始從test.txt主題讀取行並生成它們connect-test，而且接收器鏈接器應該開始從主題讀取消息connect-test 並將它們寫入文件test.sink.txt。咱們能夠經過檢查輸出文件的內容來驗證數據是否已經過整個管道傳遞：

 

1 2 3

> more test.sink.txt foo bar

 

請注意，數據存儲在Kafka主題中connect-test，所以咱們還能夠運行控制檯使用者來查看主題中的數據（或使用自定義使用者代碼來處理它）：

 

1 2 3 4

> bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic connect-test --from-beginning {"schema":{"type":"string","optional":false},"payload":"foo"} {"schema":{"type":"string","optional":false},"payload":"bar"} ...

 

鏈接器繼續處理數據，所以咱們能夠將數據添加到文件中並看到它在管道中移動：

 

1	> echo Another line>> test.txt

 

您應該看到該行出如今控制檯使用者輸出和接收器文件中。

第8步：使用Kafka Streams處理數據

Kafka Streams是一個客戶端庫，用於構建任務關鍵型實時應用程序和微服務，其中輸入和/或輸出數據存儲在Kafka集羣中。Kafka Streams結合了在客戶端編寫和部署標準Java和Scala應用程序的簡單性以及Kafka服務器端集羣技術的優點，使這些應用程序具備高度可擴展性，彈性，容錯性，分佈式等等。本快速入門示例將演示如何運行在此庫中編碼的流應用程序。