kafka安裝啟動基本測試 · ELK中文手冊

## 前提 * JDK已經安裝好，建議JDK8。如果JDK7，可能無法正常使用G1 collector * 假設您已經安裝好zookeeper集群并啟動，如果沒有，請參考[《zookeeper的安裝》](http://www.zimug.com/2016/02/20/zookeeper%E7%9A%84%E5%AE%89%E8%A3%85/) ## 單機單點安裝測試 ### 第一步：下載解壓下載地址：http://kafka.apache.org/downloads.html ``` > tar -xzf kafka_2.11-0.10.0.0.tgz > cd kafka_2.11-0.10.0.0 ``` ### 第二步：啟動kafka-server ``` > bin/kafka-server-start.sh config/server.properties ``` 關于server.properties配置文件請參考[《server.properties配置文件參數說明》](http://www.hmoore.net/hanxt/elk/159233) ### 第三步：新建一個topic 創建一個名為“test”的Topic，只有一個分區和一個備份： ``` > bin/kafka-topics.sh --create \ --zookeeper localhost:2181 \ --replication-factor 1 --partitions 1 \ --topic test ``` 創建好之后，可以通過運行以下命令，查看已創建的topic信息： ``` > bin/kafka-topics.sh --list --zookeeper localhost:2181 test ``` ### 第四步：發送消息 Kafka提供了一個命令行的工具，可以從輸入文件或者命令行中讀取消息并發送給Kafka集群。每一行是一條消息。運行producer（生產者）,然后在控制臺輸入幾條消息到服務器。 ``` > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test This is a message This is another message ``` ### 第五步：消費消息 Kafka也提供了一個消費消息的命令行工具，將存儲的信息輸出出來。 ``` > bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning This is a message This is another message ``` 如果你有2臺不同的終端上運行上述命令，那么當你在運行生產者時，消費者就能消費到生產者發送的消息。所有的命令行工具有很多的選項，你可以查看文檔來了解更多的功能。 ## 單機多點安裝 ### 第六步：設置有多個broker的集群（單機）到目前，我們只是單一的運行一個broker,，沒什么意思。對于Kafka,一個broker僅僅只是一個集群的大小, 所有讓我們多設幾個broker.**注意：broker.id從0開始遞增，每個broker實例必須唯一** 首先為每個broker創建一個配置文件: ``` > cp config/server.properties config/server-1.properties > cp config/server.properties config/server-2.properties ``` 現在編輯這些新建的文件，設置以下屬性： ``` config/server-1.properties: broker.id=1 port=9093 log.dir=/tmp/kafka-logs-1 config/server-2.properties: broker.id=2 port=9094 log.dir=/tmp/kafka-logs-2 ``` broker.id是集群中每個節點的唯一永久的名稱，我們修改端口和日志分區是因為我們現在在**同一臺機器上**運行，我們要防止broker改寫同一端口上注冊的數據。我們已經運行了zookeeper和剛才的一個kafka節點，所有我們只需要在啟動2個新的kafka節點。 ``` > bin/kafka-server-start.sh config/server-1.properties & ... > bin/kafka-server-start.sh config/server-2.properties & ... ``` 現在，我們創建一個新topic，把備份設置為：3 ``` > bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic ``` 好了，現在我們已經有了一個集群了，我們怎么知道每個broker在做什么呢？運行命令“describe topics” ``` > 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"：備份節點的集合，也就是活著的節點集合。我們運行這個命令，看看一開始我們創建的那個節點。 ``` > 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 ``` 沒有驚喜，原始主題沒有Replicas，所以是0。讓我們來發布一些信息在新的topic上： ``` > bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic ... my test message 1 my test message 2 ``` 現在，消費這些消息。 ``` > bin/kafka-console-consumer.sh --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ``` 我們要測試集群的容錯，kill掉leader，Broker1作為當前的leader，也就是kill掉Broker1。 ``` > ps | grep server-1.properties 7564 ttys002 0:15.91 /System/Library/Frameworks/JavaVM.framework/Versions/1.6/Home/bin/java... > kill -9 7564 ``` 備份節點之一成為新的leader，而broker1已經不在同步備份集合里了。 ``` > 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 --zookeeper localhost:2181 --from-beginning --topic my-replicated-topic ... my test message 1 my test message 2 ```