轉自：https://blog.csdn.net/jiankunking/article/details/67636487

基於Logstash跑通Kafka還是需要注意很多東西，最重要的就是理解Kafka的原理。

Logstash工作原理

由於Kafka採用解耦的設計思想，並非原始的釋出訂閱，生產者負責產生訊息，直接推送給消費者。而是在中間加入持久化層——broker,生產者把資料存放在broker中，消費者從broker中取資料。這樣就帶來了幾個好處:

• 1 生產者的負載與消費者的負載解耦
• 2 消費者按照自己的能力fetch資料
• 3 消費者可以自定義消費的數量

另外，由於broker採用了主題topic-->分割槽的思想，使得某個分割槽內部的順序可以保證有序性，但是分割槽間的資料不保證有序性。這樣，消費者可以以分割槽為單位，自定義讀取的位置——offset。

Kafka採用zookeeper作為管理，記錄了producer到broker的資訊，以及consumer與broker中partition的對應關係。因此，生產者可以直接把資料傳遞給broker，broker通過zookeeper進行leader-->followers的選舉管理；消費者通過zookeeper儲存讀取的位置offset以及讀取的topic的partition分割槽資訊。

由於上面的架構設計，使得生產者與broker相連；消費者與zookeeper相連。有了這樣的對應關係，就容易部署logstash-->kafka-->logstash的方案了。

接下來，按照下面的步驟就可以實現logstash與kafka的對接了。

啟動kafka

啟動zookeeper：

$zookeeper/bin/zkServer.sh start

啟動kafka：

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

建立主題

建立主題:

$kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --create --topic hello --replication-factor 1 --partitions 1

檢視主題:

$kafka/bin/kafka-topics.sh --zookeeper 127.0.0.1:2181 --describe
 

測試環境

執行生產者指令碼:

$kafka/bin/kafka-console-producer.sh --broker-list 10.0.67.101:9092 --topic hello

執行消費者指令碼，檢視是否寫入:

$kafka/bin/kafka-console-consumer.sh --zookeeper 127.0.0.1:2181 --from-beginning --topic hello

輸入測試

input{
    stdin{}
}
output{
    kafka{
        topic_id => "hello"
        bootstrap_servers => "192.168.0.4:9092" # kafka的地址
        batch_size => 5
    }
    stdout{
        codec => rubydebug
    }
}

讀取測試

logstash配置檔案:

input{
    kafka {
        codec => "plain"
        group_id => "logstash1"
        auto_offset_reset => "smallest"
        reset_beginning => true
        topic_id => "hello"
        #white_list => ["hello"]
        #black_list => nil
        zk_connect => "192.168.0.5:2181" # zookeeper的地址
   }

}
output{
    stdout{
        codec => rubydebug
    }
}