点击上方蓝色字体，选择“标星公众号”javascript
优质文章，第一时间送达php
工做流程
filebeat收集须要提取的日志文件，将日志文件转存到kafka集群中，logstash处理kafka日志，格式化处理，并将日志输出到elasticsearch中，前台页面经过kibana展现日志。css
使用kafka集群作缓存层，而不是直接将filebeat收集到的日志信息写入logstash，让总体结构更健壮，减小网络环境，致使数据丢失。filebeat负责将收集到的数据写入kafka，logstash取出数据并处理。前端
硬件条件支持
一共使用了4台服务器：
每台服务器都须要安装jdk，配置环境变量。
修改全局配置文件，做用于全部用户：java
sudo vi /etc/profileexport JAVA_HOME=JDK安装路径export PATH=$JAVA_HOME/bin:$PATH
系统调优
vim /etc/sysctl.conffs.file-max=65536vm.max_map_count = 262144
vim /etc/security/limits.conf* soft nofile 65535* hard nofile 131072* soft nproc 2048* hard nproc 4096
使用的软件版本
应用安装
kafka、zookeeper集群搭建
在10.16.10.11三、10.16.10.11四、10.16.8.187服务器中搭建kafka集群，关闭防火墙node
关闭防火墙命令：nginx
systemctl stop firewalld
查看防火墙状态：web
systemctl status firewalld
zookeeper搭建
本次直接使用kafka自带的zookeeper，不须要单独下载zookeeper面试
解压安装包到/usr/local/kafka目录下shell
vim config/zookeeper.properties
修改配置内容：
clientPort=2181maxClientCnxns=100tickTime=2000initLimit=10syncLimit=5dataDir=/usr/local/kafka/zookeeper/datadataLogDir=/usr/local/kafka/zookeeper/logserver.1=10.16.10.113:12888:13888server.2=10.16.10.114:12888:13888server.3=10.16.8.187:12888:13888
注意：dataDir、dataLogDir文件目录须要手动建立。
三台服务器配置内容一致，须要在dataDir目录下建立myid文件，文件的内容必须与zookeeper.properties中的编号保持一致。
kafka搭建
vim config/server.properties
修改配置内容：
broker.id=1prot = 9092host.name = 10.16.10.113num.network.threads=3num.io.threads=8socket.send.buffer.bytes=102400socket.receive.buffer.bytes=102400socket.request.max.bytes=104857600log.dirs=/usr/local/kafka-logsnum.partitions=16num.recovery.threads.per.data.dir=1offsets.topic.replication.factor=1transaction.state.log.replication.factor=1transaction.state.log.min.isr=1log.retention.hours=168log.segment.bytes=1073741824log.retention.check.interval.ms=300000zookeeper.connect=10.16.10.113:2181,10.16.10.114:2181,10.16.8.187:2181zookeeper.connection.timeout.ms=6000group.initial.rebalance.delay.ms=0
注意:每台服务器除broker.id 和 host.name 两个属性须要修改以外，其余属性保持一致。
验证
启动zookeeper
nohup sh zookeeper-server-start ../config/zookeeper.properties &
启动kafka
nohup sh kafka-server-start ../config/server.properties &
建立topic
/usr/local/kafka/bin/kafka-topics.sh --create --zookeeper 10.16.10.113:2181,10.16.10.114:2181,10.16.8.187:2181 --replication-factor 1 --partitions 2 --topic testtopic
查看topic
/usr/local/kafka/bin/kafka-topics.sh --zookeeper 10.16.10.113:2181,10.16.10.114:2181,10.16.8.187:2181 --list
写入消息:
命令：
/usr/local/kafka/bin/kafka-console-producer.sh --broker-list 10.16.10.113:9092 --topic testtopic
消费消息：
命令：
/usr/local/kafka/bin/kafka-console-consumer.sh --bootstrap-server 10.16.10.113:9092 --from-beginning --topic testtopic
能正常的写入消息、消费消息，kafka集群完成。
ELK 搭建
在10.16.10.11三、10.16.10.11四、10.16.3.165搭建elasticsearch,，注意关闭防火墙，修改系统的配置，注意es的启动必须是非root用户启动。
安装elasticsearch
解压文件到/usr/local/目录下,以10.16.3.165为主节点master
vim elasticsearch/config/elasticsearch.yml
修改配置：
###配置解释 cluster.name 集群名称 node.name 节点主机名 node.master 是否参与主节点竞选 node.data:true 指定该节点是否存储索引数据，默认为true。本例没配置，全部节点都存储
包括主节点
# discovery.zen.ping.unicast.hosts 配置上elasticsearch 集群除本机外其余机器# cluster.initial_master_nodes 引导启动集群的机器IP或者主机名# http.port http端口，kibana中会用到 。# transport.tcp.port 设置节点间交互的tcp端口，默认是9300。cluster.name: elkmasternode.name: 10.16.3.165node.master: truepath.logs: /usr/local/data/log/network.host: 10.16.3.165http.port: 9200discovery.zen.ping.unicast.hosts: ["10.16.10.113","10.16.10.114"]cluster.initial_master_nodes: ["10.16.3.165"]
注意：其余两台服务器，做为solver,须要修改cluster.name、node.name、network.host为自身的配置node.naster: false。最后两个属性根据服务器内容进行修改。
安装kibana
在10.16.3.165服务器上安装kibana
解压文件到/usr/local/目录下
vim kibana/config/kibana.yml
修改配置：
# i18n.locale: "zh-CN" web界面中文# server.port 监听端口server.port: 5601server.host: "10.16.3.165"elasticsearch.hosts: ["http://10.16.3.165:9200"]i18n.locale: "zh-CN"
验证
启动以前切换为非root用户。
启动elasticsearch:
命令：
nohup sh elasticsearch &# 正确的用法是下面/bin/elasticsearch -d
启动kibana
命令：
nohup sh kibana &
三台服务器访问地址ip:9200，出现如图结果说明elasticsearch启动成功
10.16.3.165访问10.16.3.165:5601，出现以下结果说明kibana启动成功
filebeat安装
在10.16.3.166服务器搭建filebeat服务
解压文件到/usr/local/目录下
vim filebeat/filebeat.yml
修改配置：
: type: log enabled: true paths: /data/home/app/domains/cpay_domain/logs/cpay-tms-gate.log : enable: true hosts: ["10.16.8.187:9092"] topic: es-tmslogs compression: gzipmax_message_bytes: 100000
注意：paths表示须要提取的日志的路径，将日志输出到kafka中，建立topic
启动filebeat:
命令：
./filebeat -e -c filebeat.yml
logstash安装
在10.16.3.165服务器搭建logstash服务
解压文件到/usr/local/目录下，建立用于本次处理的配置文件logstashfortms.conf
vim logstash/config/logstashfortms.conf
配置内容：
input{ kafka{ bootstrap_servers => "10.16.10.113:9092,10.16.10.114:9092,10.16.8.187:9092" topics => ["es-tmslogs"] codec => json }}output{ elasticsearch { hosts => ["10.16.3.165:9200"] index => "logstash-%{+YYYY.MM.dd}" }}
启动logstash:
命令：
nohup sh logstash -f ../config/logesforcpay.conf &
kibana页面操做
安装完成之后访问10.16.3.165:5601页面，选择红框中的按钮
进入索引建立模式
建立索引以后，点击红框内按钮，便可展现日志的信息,服务搭建完成。
喜欢本文的朋友们，欢迎长按下图关注订阅号成猿之路，收看更多精彩内容！
推荐阅读：
我知道你 “在看”
本文分享自微信公众号 - 成猿之路（softwareload）。
若有侵权，请联系 support@oschina.cn 删除。
本文参与“OSC源创计划”，欢迎正在阅读的你也加入，一块儿分享。