背景：
ELK由Elasticsearch、Logstash和Kibana三部分组件组成；
Logstash: logstash server端用来搜集日志；
Elasticsearch: 存储各类日志；
Kibana: web化接口用作查寻和可视化日志；
Logstash Forwarder: logstash client端用来通过lumberjack 网络协议发送日志到logstash server；
1.安装JDK
1.1下载地址
本人安装jdk-8u131-linux-x64。地址http://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html
下载完成后，在CentOS服务器上解压缩：tar xvzf jdk-8u131-linux-x64.tar.gz -C /usr/lib/jvm/
1.2配置环境变量
第一种：
vim ~/.bashrc，再文档最下部分添加：
JAVA_HOME=/usr/lib/jvm/jdk1.8.0_131
JRE_HOME=$JAVA_HOME/jre
PATH=$PATH:$JAVA_HOME/bin:/sbin:/usr/bin:/usr/sbin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export JRE_HOME
export PATH
export CLASSPATH
执行：source ~/.bashrc，使bashrc文件生效。执行java -version和java，有相应数据则安装成功。
第二种（笔者采用的第二种）：
vim /etc/profile，再文档最下部分添加：
JAVA_HOME=/usr/lib/jvm/jdk1.8.0_131
JRE_HOME=$JAVA_HOME/jre
PATH=$PATH:$JAVA_HOME/bin:/sbin:/usr/bin:/usr/sbin
CLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jar
export JAVA_HOME
export JRE_HOME
export PATH
export CLASSPATH
执行：source /etc/profile，使profile文件生效。执行java -version和java，有相应数据则安装成功。
2.安装ELK
2.1下载ELK
考虑到不是所有的服务器都具备网络环境，笔者这里自行下载。地址https://www.elastic.co/downloads/
下载完成之后，FTP上传到服务器上，笔者放到/elk文件夹下，依次解压缩，如图所示，由上到下分别为解压缩后的elasticsearch、kibana、logstash和3者的安装包。。
2.2启动elasticsearch
2.2.1修改elasticsearch.yml内容
进入elasticsearch/config文件夹，修改elasticsearch.yml文件，修改增加以下内容：
path.data: /elk/elasticsearch/data
path.logs: /elk/elasticsearch/logs
network.host: 10.8.120.37
http.port: 9200
http.cors.enabled: true
http.cors.allow-origin: "*"
2.2.2新建用户
注意：elasticsearch不能再root用户下启动，应新建普通用户如“elk”，并将总目录elk的操作权限赋给elk用户：
useradd elk
passwd elk 12345679
chown -R elk:12345679 /elk
2.2.3进入普通用户启动elasticsearch
su elk
cd /elk/elasticsearch
./bin/elasticsearch -d
浏览器输入http://ip:9200，当有以下效果时，说明elasticsearch启动成功！
启动过程中的一些错误：
1.启动报错access denied (javax.management.MBeanTrustPermission register)
方法：修改jre/lib/security/java.policy文件，在grant尾部加入permission javax.management.MBeanTrustPermission "register";
2.启动报错max virtual memory areas vm.max_map_count [65530] is too low, increase to at least [262144]
方法1：root身份下->sudo sysctl -w vm.max_map_count=262144，sysctl -a|grep vm.max_map_count，可以临时修改
方法2：root身份下->vm.max_map_count=262144直接写到/etc/sysctl.conf中,然后执行sysctl -p，达到永久修改
3.启动报错max file descriptors [65535] for elasticsearch process is too low, increase to at least [65536]
方法1：root身份下->vi /etc/security/limits.conf，* soft nofile 65536，* hard nofile 131072，* soft nproc 2048，* hard nproc 4096
2.3启动logstash
编写配置文件（名字和位置自定义即可，启动时需要指定），笔者放在config目录下，取名为screen_to_es.conf，内容如下
input {
stdin{
}
}
output {
elasticsearch {
hosts => "10.8.120.37:9200"
index => "logstash-%{+YYYY.MM.dd}"
}
启动logstash,-e:意即执行,--config或-f:意即文件执行。
cd /elk/logstash
./bin/logstash -f config/screen_to_es.conf
之后在屏幕上输入的内容将在kibana界面展示出来，具体见下文Kibana使用教程。
2.3.1 Logstash联动Kafka
由logstash输入，发送至kafka。建立logstash_to_kafka.conf，内容如下
input{
stdin{}
}
output{
kafka{
topic_id => "test"
bootstrap_servers => "10.8.120.25:9092" # kafka的地址
batch_size => 5
}
stdout{
codec => rubydebug
}
}
启动logstash
cd /elk/logstash
./bin/logstash -f config/logstash_to_kafka.conf
利用logstash从kafka读数据。建立kafka_to_logstash.conf，内容如下
input{
kafka {
codec => "plain"
group_id => "logstash"
auto_offset_reset => "smallest"
reset_beginning => true
topic_id => "test"
zk_connect => "10.8.120.25:2181" # zookeeper的地址
}
}
output{
elasticsearch {
hosts => "10.8.120.37:9200"
index => "logstash-%{+YYYY.MM.dd}"
}
}
启动logstash
cd /elk/logstash
./bin/logstash -f config/kafka_to_logstash.conf
2.3.2 Logstash联动Kafka
这里有个坑，这里有个坑，这里有个坑，笔者跳的很深！
logstash从kafka读数据时，跟 版本有很大关系，具体如下
所以当版不对应时，建议换方案实施。笔者用的是kafka_2.9.1-0.8.2.1和logstash-5.4.1，于是带来了深深的问题，因为kafka在生产系统，所以只能降logstash的版本，最终降为logstash-2.3.4。不同版本从kafka取消息的配置文件是不同的，这里笔者分别给出5.xx和两2.xx两个版本的配置文件内容。
logstash-5.4.1版本配置
input{
kafka {
bootstrap_servers => ["10.8.120.25:9092"]
topics => ["test"]
group_id => "logstash"
codec => "json"
consumer_threads =>2
decorate_events => true
}
}
output {
elasticsearch {
hosts => ["10.8.120.37:9200"]
index => "logstash-%{+YYYY.MM.dd}"
}
}
logstash-2.3.4版本配置
input{
kafka {
codec => "plain"
group_id => "logstash"
auto_offset_reset => "smallest"
reset_beginning => true
topic_id => "test"
zk_connect => "10.8.120.25:2181" # zookeeper的地址
}
}
output{
elasticsearch {
hosts => "10.8.120.37:9200"
index => "logstash-%{+YYYY.MM.dd}"
}
}
1对多：如果一个logstash同时取多个topic，运行logstash时，以config为条件运行即可，在config目录下写多个.conf文件，logstash会自动将之融合成一个大的条件。
./bin/logstash -f config
多对1：如果多个logstash同时取一个topic时，需要明确的是topic的一个分区只能被一个消费者组消费，所以对topic指向不同的分区即可。
2.4启动kibana
进入/elk/kibana/config目录，修改kibana.yml文件
server.port: 5601
server.host: "10.8.120.37"
kibana.index: ".kibana"
elasticsearch.url: "http://10.8.120.37:9200"
启动kibaba：
cd /elk/kibana
./bin/kibana
浏览器输入http://ip:5601，当有以下效果时说明kibana启动成功
3.在Kibana中查看日志
1.输入http://ip:5601进入kibana主界面。
2.进入Management管理页，点击Index Patterns索引模式。
3.默认索引为“logstash-*”，如果建立了自己的索引，可以在此输入，然后回车建立。
4.笔者这里用的默认索引，建立后如果出现这个界面，则说明你的日志收集成功了（如果未出现这个页面，请自行找一下问题的出处，logstash-elasticsearch）。
5.此时，进入Discover发现页面，即可找到自己的日志。