這篇文章主要介紹windows下如何實(shí)現(xiàn)kafka+ELK的日志系統(tǒng)，文中介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們一定要看完！
目前成都創(chuàng)新互聯(lián)已為成百上千的企業(yè)提供了網(wǎng)站建設(shè)、域名、網(wǎng)絡(luò)空間、網(wǎng)站托管、企業(yè)網(wǎng)站設(shè)計(jì)、杭州網(wǎng)站維護(hù)等服務(wù)，公司將堅(jiān)持客戶導(dǎo)向、應(yīng)用為本的策略，正道將秉承"和諧、參與、激情"的文化，與客戶和合作伙伴齊心協(xié)力一起成長，共同發(fā)展。用到的軟件：zookeeper、kafka、logstash（6.3.2版本）、ES（6.3.2版本）、Kibana（6.3.2版本）。具體安裝步驟不在此說明，基本都是下載解壓，改一下配置文件，即可使用。（以下所述均在Windows下）
1、zookeeper：
kafka中自帶zookeeper，可以不用裝zookeeper，如果想自己另裝，需配置環(huán)境變量，如下：
ZOOKEEPER_HOME => D:\nomalAPP\zookeeper-3.4.13
path 里面加入 %ZOOKEEPER_HOME%\bin
如果配置好以后，在cmd里運(yùn)行zkserver報(bào)找不到j(luò)ava錯(cuò)誤的話，可能是java環(huán)境變量的位置放置有問題，可以將path里面配置的java環(huán)境位置移到最前面。
2、kafka：
kafka如果啟動(dòng)報(bào)找不到j(luò)ava的錯(cuò)誤，原因在于kafka-run-class.bat的179行，找到里面的 %CLASSPATH% ，將其用雙引號(hào)括起來。即改為：set COMMAND=%JAVA% %KAFKA_HEAP_OPTS% %KAFKA_JVM_PERFORMANCE_OPTS% %KAFKA_JMX_OPTS% %KAFKA_LOG4J_OPTS% -cp "%CLASSPATH%" %KAFKA_OPTS% %*
kafka的配置文件server.properties里面要修改的：
日志所放置的目錄（可以選擇用默認(rèn)的）：log.dirs=D:/nomalAPP/kafka_2.12-2.0.0/kafka-logs，
連接zookeeper的ip和端口：zookeeper.connect=localhost:2181
其他的均可用默認(rèn)配置。
啟動(dòng)命令：cmd鎖定安裝目錄，然后 .\bin\windows\kafka-server-start.bat .\config\server.properties
3、logstash：
在config目錄下，logstash.yml里面，如果想啟動(dòng)多個(gè)，在里面可以配置 http.port=9600-9700，如果沒有配置這一項(xiàng)，會(huì)使用默認(rèn)的9600端口。
在config目錄下，增加一個(gè)logstash.conf的配置文件，用來配置數(shù)據(jù)源，和數(shù)據(jù)過濾，數(shù)據(jù)輸出的位置，如下：
重慶分公司
重慶分公司
input {
# ==> 用kafka的日志作為數(shù)據(jù)源
kafka{
····#（kafka的IP和端口）
bootstrap_servers => "10.8.22.15:9092"
# （在有多個(gè)相同類型的數(shù)據(jù)源時(shí)，需要配置）
client_id => "test1"
# （消費(fèi)者分組，可以通過組 ID 去指定，不同的組之間消費(fèi)是相互不受影響的，相互隔離）
group_id => "test1"
auto_offset_reset => "latest"
# （消費(fèi)者線程個(gè)數(shù)）
consumer_threads => 5
# （在輸出消息的時(shí)候會(huì)輸出自身的信息包括:消費(fèi)消息的大小， topic 來源以及 consumer 的 group 信息）
decorate_events => true
# （主題）
topics => ["bas-binding-topic"]
# （用于ES索引）
type => "bas-binding-topic"
# （數(shù)據(jù)格式）
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插件負(fù)責(zé)過濾解析input讀取的數(shù)據(jù)，可以用grok插件正則解析數(shù)據(jù)，
date插件解析日期，json插件解析json等等
}
output {
# 輸出到ES
if "bas-binding-topic" in [tags] {
elasticsearch{
# ES的IP和端口
hosts => ["localhost:9201"]
# 索引名稱：主題+時(shí)間
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
}
}
}啟動(dòng)logstash的命令： .\bin\logstash.bat -f .\config\logstash.conf
4、ES：
在config目錄下，elasticsearch.yml配置文件中，ES的默認(rèn)端口為9200，可以通過http.port=9201來修改
啟動(dòng)命令： .\bin\elasticsearch.bat
啟動(dòng)后，在瀏覽器訪問： http://localhost:9201,
如果出現(xiàn)了一些信息，表示啟動(dòng)成功。
5、Kibana：
在config目錄下，kibana.yml配置文件中，通過elasticsearch.url: "http://localhost:9201" 來配置地址
啟動(dòng)命令： .\bin\kibana.bat
訪問 "http://localhost:9201" ，如果出現(xiàn)需要輸入用戶名密碼的界面，表示ES啟動(dòng)失敗，如果直接顯示Kibana界面，則啟動(dòng)成功。
logstash數(shù)據(jù)輸出到ES時(shí)，會(huì)選擇ES默認(rèn)的映射來解析數(shù)據(jù)。如果覺得默認(rèn)映射不滿足使用條件，可以自定模板：
借助postman工具創(chuàng)建ES模板，如下：
收到此響應(yīng)表示創(chuàng)建成功。然后可以通過GET請(qǐng)求查詢你剛才創(chuàng)建的模板，通過DELETE請(qǐng)求刪除剛才創(chuàng)建的模板。
向kafka主題中發(fā)送日志信息，發(fā)送的信息會(huì)在cmd窗口顯示，如下：
同時(shí)，在Kibana界面會(huì)看到相應(yīng)的索引，索引名稱就是在logstash.conf的輸出中配置的，如下：
Visualize：可以選擇自己想要的索引去進(jìn)行圖形分析，效果如下：
Dashboard：將做過的圖整合到儀表盤，效果如下：
另外，可以設(shè)置自動(dòng)刷新，即當(dāng)有新的數(shù)據(jù)發(fā)送到kafka時(shí)，所做的圖會(huì)自動(dòng)根據(jù)主題來進(jìn)行刷新。
項(xiàng)目中，可以通過logback來講日志輸出到kafka，具體配置如下：
