ELK结合kafka环境搭建，简单直接，有手就行

科技 08-30 来源： Java开发技术分享

ELK搭建全流程

环境配置要求

组件安装包	版本	描述
Java运行环境	jdk1.8
elasticsearch	8.2.0	日志存储
IK分词器	8.2.0	分词工具
filebeat	8.2.0	日志采集
logstash	8.2.0	分析处理过滤
metricbeat	8.2.0	metric采集
kibana	8.2.0	图形化展示

注：elasticsearch，Ik，filebeat，logstash，metricbeat，kibana必须版本一致。

基本流程：

Linux环境配置调优

## 修改3节点hosts文件
vim /etc/hosts
192.168.11.1 wchi1
192.168.11.2 wchi2
192.168.11.3 wchi3

## 挂载光驱
mount /dev/cdrom /media

## 安装sudo
yum install sudo

## 禁用swap
sudo swapoff -a

## 修改linux内核参数
sudo vim /etc/security/limits.conf

## 添加如下内容:
* soft nofile 655360
* hard nofile 655360
* soft memlock unlimited
* hard memlock unlimited

## 修改虚拟内存空间及swap使用率
cd /proc/sys/vm

## 设置如下:
sysctl -w vm.max_map_count=655360
sysctl -w vm.swappiness=1

## 查看
cat /proc/sys/vm/swappiness
cat /proc/sys/vm/max_map_count

#设置永久有效需要把配置信息添加到sysctl.conf文件中去 
## 命令：
vim /etc/sysctl.conf

## 添加：
vm.max_map_count=655360
vm.swappiness=1

## 执行命令：使配置生效 
    sudo sysctl -p

## 修改创建本地线程数
sudo vim /etc/security/limits.d/20-nproc.conf

## 修改为：
* soft nproc 4096

Elasticsearch环境搭建

## 三个节点解压elasticsearch-8.2.0.tar.gz
tar -zxvf elasticsearch-7.15.0-linux-x86_64.tar.gz -C /usr/local/

## 修改配置文件：
vim elasticsearch-7.15.0/config/elasticsearch.yml

## elasticsearch.yml 配置
cluster.name: es_log_cluster
## es-node-2  es-node-3 不同节点名称不同
node.name: es-node-1
#### es数据存放位置
path.data: /usr/local/elasticsearch-8.2.0/data
#### es日志存放位置
path.logs: /usr/local/elasticsearch-8.2.0/logs
#### 锁内存,强制占用(类似oracle的锁内存)保证es启动正常
bootstrap.memory_lock: true
#### network.host不同节点IP对应 （对外发布IP）
network.host: 192.168.142.128
#### http端口
http.port: 9200
#### 集群列表
discovery.seed_hosts: ["192.168.142.128","192.168.142.129","192.168.142.131"]
#### 可作为主节点列表
cluster.initial_master_nodes: ["192.168.142.128","192.168.142.129","192.168.142.131"]
## 配置认证
#http.cors.allow-headers: Authorization,X-Requested-With,Content-Length,Content-Type
xpack.security.enabled: false
xpack.security.transport.ssl.enabled: true
xpack.security.transport.ssl.verification_mode: certificate
xpack.security.transport.ssl.keystore.path: /usr/local/elasticsearch-8.2.0/config/elastic-certificates.p12
xpack.security.transport.ssl.truststore.path: /usr/local/elasticsearch-8.2.0/config/elastic-certificates.p12
http.cors.enabled: true
http.cors.allow-origin: "*"

##############生成认证####################
## 直接执行：
/usr/local/elasticsearch-7.15.0/bin/elasticsearch-certutil ca
/usr/local/elasticsearch-7.15.0/bin/elasticsearch-certutil cert --ca /usr/local/elasticsearch-8.2.0/elastic-stack-ca.p12

## 将生成的证书移动到config目录
mv elastic-certificates.p12 /usr/local/elasticsearch-7.15.0/config/

## 将生成的文件复制到所有节点对应路径/usr/local/elasticsearch-8.2.0/config
## 如果一下命令报错，直接下载上传文件过去也一样
cd /usr/local/elasticsearch-7.15.0/config/
scp elastic-certificates.p12 192.168.11.62:/usr/local/elasticsearch-7.15.0/config
scp elastic-certificates.p12 192.168.11.62:/usr/local/elasticsearch-7.15.0/config

## 创建文件夹：
mkdir /usr/local/elasticsearch-7.15.0/data
mkdir /usr/local/elasticsearch-7.15.0/logs

## 赋权
chown -R wchi:wchi /usr/local/elasticsearch-7.15.0/

## 测试启动 注意：启动es不能用root用户，需要用普通用户
su wchi
/usr/local/elasticsearch-7.15.0/bin/elasticsearch

## ES启动情况下设置密码，ES集群开启xpack的时候只要有一个节点设置密码即可，否则不能访问，全部设置123456
cd /usr/local/elasticsearch-7.15.0/bin
elasticsearch-setup-passwords interactive

## 集群测试：
curl -u elastic:123456 192.168.11.61:9200

IK分词器安装

####################################  安装ik分词器  ############################
## ik分词器：
## 安装elasticsearch-ik分词器
https://github.com/medcl/elasticsearch-analysis-ik
## 下载地址：
https://github.com/medcl/elasticsearch-analysis-ik/releases
## 8.2.0版本：
https://github.com/medcl/elasticsearch-analysis-ik/releases/download/v7.15.0/elasticsearch-analysis-ik-8.2.0.zip

## 创建文件夹：
mkdir -p /usr/local/elasticsearch-7.15.0/plugins/ik/

## 进行解压到刚创建的/usr/local/elasticsearch-8.2.0/plugins/ik/目录：
unzip -d /usr/local/elasticsearch-7.15.0/plugins/ik/ elasticsearch-analysis-ik-7.15.0.zip

## 查看是否ok
cd /usr/local/elasticsearch-7.15.0/plugins/ik/
## 重新复权
chown -R wchi:wchi /usr/local/elasticsearch-7.15.0/

## 重新启动ES节点，显示如下信息代表加载ik分词器成功
[es-node01] loaded plugin [analysis-ik]

metricbeat安装

##################################   metricbeat   ################################

## 解压metricbeat
tar -zxvf metricbeat-7.15.0-linux-x86_64.tar.gz -C /usr/local/
mv metricbeat-8.2.0-linux-x86_64 metricbeat-7.15.0

## 在 Metricbeat 中启用并配置 Elasticsearch x-pack 模块
cd /usr/local/metricbeat-7.15.0
metricbeat modules enable elasticsearch-xpack

## 修改配置：
cd /usr/local/metricbeat-7.15.0/modules.d
chmod 744 elasticsearch-xpack.yml
vim /usr/local/metricbeat-7.15.0/modules.d/elasticsearch-xpack.yml

- module: elasticsearch
  xpack.enabled: true
  period: 10s
  hosts: ["http://192.168.11.61:9200","http://192.168.11.62:9200","http://192.168.11.63:9200"]
  username: "elastic"
  password: "123456"

## 配置metricbeat.yml，配置 Metricbeat 以发送至监测集群
vim /usr/local/metricbeat-7.15.0/metricbeat.yml

output.elasticsearch:
  hosts: ["http://192.168.11.61:9200","http://192.168.11.62:9200","http://192.168.11.63:9200"] 
  protocol: "https"
  username: "elastic"
  password: "123456"

setup.kibana:
  host: "192.168.11.61:5601"
  username: "elastic"
  password: "123456"

## 整个目录赋权
chown -R wchi:wchi /usr/local/metricbeat-7.15.0/

## 加载索引模板供 Kibana 可视化
./metricbeat setup -e
##PS: 执行成功后，会有如上的显示：“kibana dashboards successfully loaded.

## 启动metricbeat
/usr/local/metricbeat-7.15.0/metricbeat -e &

## 查看进程
ps -ef | grep metricbeat

filebeat安装

## 解压filebeat
tar -zxvf filebeat-7.15.0-linux-x86_64.tar.gz -C /usr/local/

##新建自己的配置文件

##简单配置，从log文件收集日志，输出到es
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/log/*.log
#setup.template.enabled: false
#setup.template.name: "mylog"
#setup.template.pattern: "mylog-*"
#setup.ilm.enabled: false
output.elasticsearch:
  hosts: ["192.168.142.128:9200"]
  username: "elastic"
  password: "123456"
  #index: "mylog"
  #enable: true

##简单配置，从log文件收集日志，输出到kafka
filebeat.inputs:
- type: log
  enabled: true
  paths:
    - /usr/local/logs/app-user-core.log
  fields:
    log_topic: appUserLog


- type: log
  enabled: true
  paths:
    - /usr/local/logs/error-user-core.log
  fields:
    log_topic: errorUserLog


- type: log
  enabled: true
  paths:
    - /usr/local/logs/trace-user-core.log
  fields:
    log_topic: traceUserLog

output.kafka:
  enabled: true
  hosts: ["192.168.142.128:9092"]
  topic: '%{[fields.log_topic]}'

##filebeat启动
./filebeat -e -c ./filebeat-kafka.yml

kibana安装

##################################   kibana   ################################

## 解压kibana
tar -zxvf kibana-7.15.0-linux-x86_64.tar.gz -C /usr/local/
## 进入kibana目录，修改配置文件
vim /usr/local/kibana-7.15.0/config/kibana.yml
## 修改配置如下：
server.host: "0.0.0.0"
server.name: "192.168.11.61"
elasticsearch.hosts: ["http://192.168.11.61:9200","http://192.168.11.62:9200","http://192.168.11.63:9200"]
elasticsearch.username: "kibana_system"
elasticsearch.password: "123456"
i18n.locale: "zh-CN"

## 赋权
chown -R baihezhuo:baihezhuo /usr/local/kibana-7.15.0/

## 前台启动：
/usr/local/kibana-7.15.0/bin/kibana -c /usr/local/kibana-7.15.0/config/kibana.yml &
## 后台启动：
nohup /usr/local/kibana-7.15.0/bin/kibana -c /usr/local/kibana-7.15.0/config/kibana.yml > /dev/null 2>&1 &

## 访问地址: (5601为kibana默认端口)
http://192.168.11.61:5601/app/kibana

netstat -tunpl | grep 5601

zookeeper环境搭建

1. 准备工作：
## 准备3个节点，要求配置好主机名称，服务器之间系统时间保持一致
## 注意 /etc/hostname 和 /etc/hosts 配置主机名称（在这个里我准备bhz125,bhz126,bhz127三节点）
## 特别注意 以下操作3个节点要同时进行操作哦！

2. 上传zk到三台服务器节点
## 注意我这里解压到/usr/local下
2.1 进行解压： tar zookeeper-3.4.6.tar.gz
## 2.2 重命名： mv zookeeper-3.4.6 zookeeper
2.3 修改环境变量： vim /etc/profile 
## 这里要添加zookeeper的全局变量
export ZOOKEEPER_HOME=/usr/local/zookeeper-3.4.6
export PATH=.:$ZOOKEEPER_HOME/bin

## 当环境有问题时暂时使用：export PATH=/usr/bin:/usr/sbin:/bin:/sbin:/usr/X11R6/bin
2.4 刷新环境变量： source /etc/profile
2.5 到zookeeper下修改配置文件： 
2.5.1 首先到指定目录： cd /usr/local/zookeeper-3.4.6/conf
2.5.2 然后复制zoo_sample.cfg文件，复制后为zoo.cfg： mv zoo_sample.cfg zoo.cfg
2.5.3 然后修改两处地方, 最后保存退出：
(1) 修改数据的dir
dataDir=/usr/local/zookeeper-3.4.6/data
(2) 修改集群地址
server.0=bhz125:2888:3888
server.1=bhz126:2888:3888
server.2=bhz127:2888:3888

2.6 增加服务器标识配置，需要2步骤，第一是创建文件夹和文件，第二是添加配置内容： 
(1) 创建文件夹： mkdir /usr/local/zookeeper-3.4.6/data
(2) 创建文件myid 路径应该创建在/usr/local/zookeeper/data下面，如下：
	vim /usr/local/zookeeper-3.4.6/data/myid
	注意这里每一台服务器的myid文件内容不同，分别修改里面的值为0，1，2；与我们之前的zoo.cfg配置文件里：server.0，server.1，server.2 顺序相对应，然后保存退出；

2.7 到此为止，Zookeeper集群环境大功告成！启动zookeeper命令
	启动路径：/usr/local/zookeeper-3.4.6/bin（也可在任意目录，因为配置了环境变量）
	执行命令：zkServer.sh start (注意这里3台机器都要进行启动，启动之后可以查看状态)
	查看状态：zkServer.sh status (在三个节点上检验zk的mode, 会看到一个leader和俩个follower)
	
## zkCli.sh 进入zookeeper客户端

根据提示命令进行操作： 
查找：ls /   ls /zookeeper
创建并赋值： create /imooc zookeeper
获取： get /imooc 
设值： set /imooc zookeeper1314 
PS1: 任意节点都可以看到zookeeper集群的数据一致性
PS2: 创建节点有俩种类型：短暂（ephemeral） 持久（persistent）, 这些小伙伴们可以查找相关资料，我们这里作为入门不做过多赘述！

开机启动zookeeper

cd /etc/rc.d/init.d/
touch zookeeper
chmod 777 zookeeper
vim zookeeper

开机启动zookeeper脚本：
#!/bin/bash

#chkconfig:2345 20 90
#description:zookeeper
#processname:zookeeper
export JAVA_HOME=/usr/local/jdk1.8
export PATH=$JAVA_HOME/bin:$PATH
case $1 in
		start) /usr/local/zookeeper-3.4.6/bin/zkServer.sh start;;
		stop) /usr/local/zookeeper-3.4.6/bin/zkServer.sh stop;;
		status) /usr/local/zookeeper-3.4.6/bin/zkServer.sh status;;
		restart) /usr/local/zookeeper-3.4.6/bin/zkServer.sh restart;;
		*)  echo "require start|stop|status|restart"  ;;
esac

开机启动配置：chkconfig zookeeper on

验证：
  chkconfig --add zookeeper
  
  chkconfig --list zookeeper
  
这个时候我们就可以用service zookeeper start/stop来启动停止zookeeper服务了
  
使用chkconfig--add zookeeper命令把zookeeper添加到开机启动里面
  
添加完成之后接这个使用chkconfig--list 来看看我们添加的zookeeper是否在里面
  
如果上面的操作都正常的话；你就可以重启你的linux服务器了

KAFKA环境搭建

kafka下载地址：http://kafka.apache.org/downloads.html

## 解压命令：
tar -zxvf kafka_2.12.tgz -C /usr/local
## 进入解压后的目录，修改server.properties文件：
vim /usr/local/kafka_2.12/config/server.properties

## 修改配置：
broker.id=0
port=9092
host.name=192.168.11.114
advertised.host.name=192.168.11.114
log.dirs=/usr/local/kafka_2.12/kafka-logs
num.partitions=2
zookeeper.connect=192.168.11.111:2181,192.168.11.112:2181,192.168.11.113:2181

## 建立日志文件夹：mkdir /usr/local/kafka/kafka-logs
##启动kafka：
/usr/local/kafka_2.12/bin/kafka-server-start.sh /usr/local/kafka_2.12/config/server.properties &

## kafka manager
## 下载kafka-manager-2.0.0.2.zip
##解压：
unzip kafka-manager-2.0.0.2.zip -d /usr/local/
## 编辑文件：
vim /usr/local/kafka-manager-2.0.0.2/conf/application.conf
## 修改内容
kafka-manager.zkhosts="192.168.11.111:2181,192.168.11.112:2181,192.168.11.113:2181"
## 启动kafka-manager：
/usr/local/kafka-manager-2.0.0.2/bin/kafka-manager &
## 默认端口为：9000
http://192.168.11.51:9000

## Kafka的shell脚本都在bin目录下。
cd /usr/local/kafka_2.12/bin
## Shell使用请参考博客地址： http://www.cnblogs.com/xiaodf/p/6093261.html

## 简单操作：
#（1）创建topic主题命令：（创建名为test的topic， 1个分区分别存放数据，数据备份总共1份）
## 注意 以后一个应用服务一个topic  ： log-app-应用服务名称
kafka-topics.sh --zookeeper 192.168.11.111:2181 --create --topic log-app-topic --partitions 1  --replication-factor 1  
## 注意 以后一个应用服务一个topic  ： log-trace-应用服务名称
kafka-topics.sh --zookeeper 192.168.11.111:2181 --create --topic log-trace-topic --partitions 1  --replication-factor 1  

## 修改分区数
kafka-topics.sh --zookeeper 192.168.11.111:2181 --alter --topic topicName --partitions 64

#（2）查看topic列表命令：
kafka-topics.sh --zookeeper 192.168.11.111:2181 --list
生产与消费消息：
#（3）kafka命令发送数据：（然后我们就可以编写数据发送出去了）
kafka-console-producer.sh --broker-list 192.168.11.51:9092 --topic log-trace-topic
#（4）kafka命令接受数据：（然后我们就可以看到消费的信息了）
kafka-console-consumer.sh --zookeeper 192.168.11.111:2181 --topic  log-trace-topic
kafka-console-consumer.sh --zookeeper 192.168.11.111:2181 --topic  log-app-topic