windows下kafka+ELK的日誌系統
用到的軟體：zookeeper、kafka、logstash（6.3.2版本）、ES（6.3.2版本）、Kibana（6.3.2版本）。具體安裝步驟不在此說明，基本都是下載解壓，改一下配置檔案，即可使用。（以下所述均在Windows下）
1、zookeeper：
kafka中自帶zookeeper，可以不用裝zookeeper，如果想自己另裝，需配置環境變數，如下：
ZOOKEEPER_HOME => D:\nomalAPP\zookeeper-3.4.13
path 裡面加入 %ZOOKEEPER_HOME%\bin
如果配置好以後，在cmd裡執行zkserver報找不到java錯誤的話，可能是java環境變數的位置放置有問題，可以將path裡面配置的java環境位置移到最前面。</b>
2、kafka：
kafka如果啟動報找不到java的錯誤，原因在於kafka-run-class.bat的179行，找到裡面的 %CLASSPATH% ，將其用雙引號括起來。即改為：set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %* </b>
kafka的配置檔案server.properties裡面要修改的：
日誌所放置的目錄（可以選擇用預設的）：log.dirs=D:/nomalAPP/kafka_2.12-2.0.0/kafka-logs，
連線zookeeper的ip和埠：zookeeper.connect=localhost:2181
其他的均可用預設配置。</b>
啟動命令：cmd鎖定安裝目錄，然後 .\bin\windows\kafka-server-start.bat .\config\server.properties
</b>
3、logstash：
在config目錄下，logstash.yml裡面，如果想啟動多個，在裡面可以配置 http.port=9600-9700，如果沒有配置這一項，會使用預設的9600埠。
在config目錄下，增加一個logstash.conf的配置檔案，用來配置資料來源，和資料過濾，資料輸出的位置，如下：
input { # ==> 用kafka的日誌作為資料來源 kafka{ ····#（kafka的IP和埠） bootstrap_servers => "10.8.22.15:9092" # （在有多個相同型別的資料來源時，需要配置） client_id => "test1" # （消費者分組，可以通過組 ID 去指定，不同的組之間消費是相互不受影響的，相互隔離） group_id => "test1" auto_offset_reset => "latest" # （消費者執行緒個數） consumer_threads => 5 # （在輸出訊息的時候會輸出自身的資訊包括:消費訊息的大小， topic 來源以及 consumer 的 group 資訊） decorate_events => true # （主題） topics => ["bas-binding-topic"] # （用於ES索引） type => "bas-binding-topic" # （資料格式） codec => "json" tags => ["bas-binding-topic"] } kafka{ bootstrap_servers => "10.8.22.15:9092" client_id => "test2" group_id => "test2" auto_offset_reset => "latest" consumer_threads => 5 #decorate_events => true topics => ["bas-cus-request-topic"] type => "bas-cus-request-topic" codec => "json" tags => ["bas-cus-request-topic"] } } filter { filter外掛負責過濾解析input讀取的資料，可以用grok外掛正則解析資料， date外掛解析日期，json外掛解析json等等 } output { # 輸出到ES if "bas-binding-topic" in [tags] { elasticsearch{ # ES的IP和埠 hosts => ["localhost:9201"] # 索引名稱：主題+時間 index => "bas-binding-topic-%{+YYYY.MM.dd+HH:mm:ss}" timeout => 300 } stdout { codec => rubydebug } } if "bas-cus-request-topic" in [tags] { elasticsearch{ hosts => ["localhost:9201"] index => "bas-cus-request-topic-%{+YYYY.MM.dd+HH:mm:ss}" timeout => 300 } stdout { codec => rubydebug } } }
啟動logstash的命令： .\bin\logstash.bat -f .\config\logstash.conf </b>
4、ES：
在config目錄下，elasticsearch.yml配置檔案中，ES的預設埠為9200，可以通過http.port=9201來修改
啟動命令： .\bin\elasticsearch.bat
啟動後，在瀏覽器訪問： http://localhost:9201,
如果出現了一些資訊，表示啟動成功。</b>
5、Kibana：
在config目錄下，kibana.yml配置檔案中，通過elasticsearch.url: "http://localhost:9201" 來配置地址
啟動命令： .\bin\kibana.bat
訪問 "http://localhost:9201" ，如果出現需要輸入使用者名稱密碼的介面，表示ES啟動失敗，如果直接顯示Kibana介面，則啟動成功。</b>
logstash資料輸出到ES時，會選擇ES預設的對映來解析資料。如果覺得預設對映不滿足使用條件，可以自定模板：
藉助postman工具建立ES模板，如下：
收到此響應表示建立成功。然後可以通過GET請求查詢你剛才建立的模板，通過DELETE請求刪除剛才建立的模板。</b>
向kafka主題中傳送日誌資訊，傳送的資訊會在cmd視窗顯示，如下：
同時，在Kibana介面會看到相應的索引，索引名稱就是在logstash.conf的輸出中配置的，如下：
Visualize：可以選擇自己想要的索引去進行圖形分析，效果如下：
Dashboard：將做過的圖整合到儀表盤，效果如下：
另外，可以設定自動重新整理，即當有新的資料傳送到kafka時，所做的圖會自動根據主題來進行重新整理。
專案中，可以通過logback來講日誌輸出到kafka，具體配置如下：
<appender name="bingdings-log-to-kafka" class="com.github.danielwegener.logback.kafka.KafkaAppender">
<encoder>
<pattern>%msg%n</pattern>
</encoder>
<!-- 指定kafka主題，消費時要用到-->
<topic>bas-binding-topic</topic>
<keyingStrategy class="com.github.danielwegener.logback.kafka.keying.NoKeyKeyingStrategy" />
<deliveryStrategy class="com.github.danielwegener.logback.kafka.delivery.AsynchronousDeliveryStrategy" />
<!-- 配置輸出到指定ip、埠的kafka -->
<producerConfig>bootstrap.servers=localhost:9092,10.8.22.13:9092</producerConfig>
<!-- this is the fallback appender if kafka is not available. -->
</appender>
<logger name = "bingdings-log-to-kafka" level="INFO" additivity = "false">
<appender-ref ref="bingdings-log-to-kafka"/>
</logger>然後在記日誌的時候，記到對應的主題，日誌就會寫到kafka相應的位置。可以通過消費者來消費主題，檢視日誌是否成功寫入，如下：
@SuppressWarnings("resource")
public static void main(String[] args) {
Properties properties = new Properties();
properties.put("bootstrap.servers", "127.0.0.1:9092");
properties.put("group.id", "group-1");
properties.put("enable.auto.commit", "false");
properties.put("auto.commit.interval.ms", "1000");
properties.put("auto.offset.reset", "earliest");
properties.put("session.timeout.ms", "30000");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer",
"org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(properties);
kafkaConsumer.subscribe(Arrays.asList("bas-cus-request-topic", "bas-binding-topic"));
while (true) {
ConsumerRecords<String, String> records = kafkaConsumer.poll(Long.MAX_VALUE);
System.err.println("+++++++++++++++++++++++++++++++++++++++++");
for (ConsumerRecord<String, String> record : records) {
System.err.println(record.offset() + ">>>>>>" + record.value());
}
System.err.println("+++++++++++++++++++++++++++++++++++++++++++++");
break;
}
}暫時想到的就這些，有建議的，歡迎評論提醒我進行補充，謝謝！