kafka快速入門（官方文檔）

第1步：下載代碼

下載 1.0.0版本並解壓縮。

> tar -xzf kafka_2.11-1.0.0.tgz
> cd kafka_2.11-1.0.0

第2步：啓動服務器

Kafka使用ZooKeeper，因此若是你尚未ZooKeeper服務器，你須要先啓動一個ZooKeeper服務器。您可使用與kafka一塊兒打包的便捷腳原本獲取快速而簡單的單節點ZooKeeper實例。

> 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服務器：

> 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」的主題：

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

咱們如今能夠看到這個主題，若是咱們運行列表主題命令：

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

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

第4步：發送一些消息

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

運行生產者，而後在控制檯輸入一些消息發送到服務器。

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

第5步：啓動一個用戶

Kafka也有一個命令行消費者，將消息轉儲到標準輸出。

> 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只是一個規模一個的集羣，因此除了開始一些borker實例以外沒有太大的變化。可是爲了獲得它的感受，讓咱們把咱們的集羣擴展到三個節點（仍然都在咱們的本地機器上）。

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

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

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

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

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

咱們已經有Zookeeper和咱們的單節點了，因此咱們只須要啓動兩個新的節點：

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

...

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

...

如今建立一個ReplicationFactor爲3的新主題：

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

好吧，如今咱們有一個集羣，咱們怎麼知道哪一個broker在作什麼？要查看運行「描述主題」命令：

> 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」是複製此分區日誌的節點列表，不管它們是不是leader，或即便他們當前處於活動狀態。
• 「isr」是一組「同步」副本。這是replicas列表的子集，當前活着並被引導到leader。

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

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

> 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上，這是咱們建立集羣時惟一的服務器。

讓咱們發表一些信息給咱們的新主題：

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

如今讓咱們消費這些消息：

> 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是做爲領導者，因此讓咱們殺了它：

> 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上使用：

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

領導已經切換到其中一個從屬節點，而且節點1再也不處於同步副本集合中：

> 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

可是，即便原先寫入的領導者失敗，這些消息仍然可用於消費：

> 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主題導出到文件。

首先，咱們將經過建立一些種子數據開始測試：

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

或在Windows上：

> echo foo> test.txt
> echo bar>> test.txt

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

> 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。咱們能夠經過檢查輸出文件的內容來驗證經過整個管道傳輸的數據：

> more test.sink.txt
foo
bar

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

> 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"}
...

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

> echo Another line>> test.txt

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

第8步：使用Kafka流來處理數據

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