当前位置：首页 > news >正文

ELK的搭建

news 文章来源：https://blog.csdn.net/yangfeipancc/article/details/144913738 2025/2/9 9:58:29

ELK

elk：elasticsearch logstatsh kibana统一日志收集系统

elasticsearch：分布式的全文索引引擎点非关系型数据库,存储所有的日志信息，主和从，最少需要2台

logstatsh：动态的从各种指定的数据源，获取数据，而且对数据进行过滤，分析，按照统一的格式发送到es

kibana：把es的数据进行展示，进行客户端可以理解的操做，汇总，分析和搜索数据的平台

ELK工作原理

1、logstash收集日志数据
2、收集到的数据按照es的格式，发送到数据库
3、在图形界面展示—kibana
4、客户端访问kibana

安装

test1 es1

test2 es2

test3 logstatsh kibana
三台主机

cd /opt
把es安装包拖到/opt目录下
apt -y install openjdk-8-jre-headless

test1和test2

dpkg -i elasticsearch
free -h 
echo 3 > /proc/sys/vm/drop_caches
#清理内存
cd /etc/elasticsearch 
cp elasticsearch.yml elasticsearch.yml.bak.2025.1
vim  elasticsearch.yml
第17行取消注释
cluster.name: my-elk
第23行
node.name: node1/node2
#test1和test2名称不能一样
node.master：true/false
#表示该节点是否为主节点，test1是true，test2为false
node.date：true
#表示该节点是否为数据节点
第57行去掉注释
network.host:0.0.0.0
#修改监听地址
第64行去掉注释
http.port:9200
#es对外提供的默认端口
transport.tcp.port:9300
#指定es集群内部的通信接口
第76行取消注释
discovery.zen.ping.unicast.hosts: ["192.168.42.50:9300", "192.168.42.51:9300"]
#集群节点通过单播的形式，通过9300端口实现内部通信
http.cors.enabled:true
#开启跨域访问
http.cors.allow-origin:"*"
#允许跨域访问的地址域名为所有
保存退出
grep -v “^#” elasticsearch.yml
systemctl restart elasticsearch
#查看日志
tail -f /var/log/syslog
#查看端口
netstat -antp | grep 9200

浏览器查看
test1
在这里插入图片描述
test2

test3

dpkg -i logstash
dpkg -i kibana
systemctl restart logstash
cd /usr/share/logstash/bin/
ln -s /usr/share/logstash/bin/logstash /usr/local/bin/
cd /erc/kibana/
vim kibana.yml
第2行取消注释
第7行server.host 0.0.0.0
第28行
elasticsearch.hosts: ["http://192.168.233.10:9200", "http://192.168.233.20:9200"]
第37行取消注释
第96行/var/log/kibana.log
第113行 zh-CN
保存退出
touch /var/log/kibana.log
chown kibana.kibana /var/log/kibana.log
systemctl restart kibana
查看日志
查看端口
cd /etc/logstash/conf.d
vim system.conf
input {file{path=>"/var/log/syslog"type=>"system"start_position=>"beginning"}
}
output {elasticsearch {hosts => ["192.168.42.60;9200","192.168.42.61:9200"]index => "system-%{+YYYY.MM.dd}"}
}
保存退出
logstash -f system.conf --path.data /opt/test1 &

在这里插入图片描述

logstash:logstash是一个重量级的工具，占用很大的内存空间，只能收集本机的服务，没有办法远程发送

filebeat结合logstash实现远程的日志收集

filebeat

filebeat是一款轻量级的日志收集工具，不依赖java环境，用来替代在机器上没有java环境的情况下进行日志收集，filebeat启动日志，只要10M左右内存
test4

vim /etc/my.cnf
general_log=ON
general_log_file=/usr/local/mysql/data/mysql_general.log
保存退出
systemctl restart mysql
apt -y install nginx
systemctl restart nginx
cd /opt
把filebeat安装包拖进来
tar -xf filebeat
mv  filebeat  /usr/local/filebeat
cd /usr/local/filebeat
vim filebeat.yml
filebeat.inputs:
- type: logenabled: truepaths:- /var/log/nginx/access.log- /var/log/nginx/error.logtags: ["nginx"]#标签，为了后续logstash进行识别的fields:service_name: 192.168.42.23_nginx#设定显示的索引名称log_type: nginxfrom: 192.168.233.21
- type: logenabled: truepaths:- /usr/local/mysql/data/mysql_general.logtags: ["mysql"]#标签，为了后续logstash进行识别的fields:service_name: 192.168.42.23_mysql#设定显示的索引名称log_type: mysqlfrom: 192.168.42.23
seccomp:
default_action: allow
syscalls:
- action: allownames:- rseq第157行和159行注释第170行和172行取消注释Ihosts: ["192.168.42.62:5044"]
logstash的端口，不是说logstash一定是5044，理论上来说可以是任意端口，默认的logstash起始端口
保存退出
cd /usr/local/filebeat
./filebeat -e -c filebeat.yml
#-e:输出的格式为标准输出
#-c:指定配置文件

test3

cd /etc/logstash/conf.d
vim test1.conf
input {beats { port => "5044"}
}
output {if "nginx" in [tags] {elasticsearch {hosts => ["192.168.42.60:9200", "192.168.42.61:9200"]index => "%{[fields][serrvice_name]}-%{+YYYY.MM.dd}"}}if "mysql" in [tags] {elasticsearch {hosts => ["192.168.42.60:9200", "192.168.42.61:9200"]index => "%{[fields][serrvice_name]}-%{+YYYY.MM.dd}"}}
}
保存退出
logstash -f test1.conf --path.data /opt/test2 &

在这里插入图片描述

zookeeper+kafka

zookeepr:开源的分布式框架协调服务

zookeeper工作机制：

1、基于观察者模式设计的分布式结构，负责存储和管理的架构当中的元信息，架构当中的应用接受观察者的监控，一旦数据有变化，通知对应的zookeeper，保存变化的信息

2、特点：

最少要有3台机器，一个领导者（leader），多个跟随者（follower）
zookeeper要有半数以上的节点存活，整个架构就可以正常工作，所以都是奇数台部署
全局数据一致
数据更新的原子性，要么都成功，要么都失败
实时性

zookeeper的数据结构：

1、统一命名服务，不是以ip来记录，可以用域名也可以是主机名来记录信息

2、统一配置管理，所有的节点信息的配置要一致

3、同一集群管理，在整个分布式的环境中，必须实时的掌握每个节点的状态，如果状态发生变化，要及时更新
nginx1/2/3

apt -y install ntpdate
ntpdate alyun.comm
cd /opt
把zookeeper托进去
tar -xf zookeeper
apt -y install java
java -version
mv zpzche-zookeeper zookeeper
cd zookeeper/conf
cp zoo_sample.cfg zoo.cfg
vim zoo.cfg
dataDir=/opt/zookeeper/data
dataLogDir=/opt/zookeeper/logs
server.1=192.168.42.20:3188:3288
server.2=192.168.42.30:3188:3288
server.3=192.168.42.40:3188:3288
保存退出
cd /opt/zookeeper
mkdir data
mkdir logs
useradd -m -s /sbin/nologin zookeeper
cd /opt
chown -R zookeeper.zookeeper /opt/zookeeper
cd zookeeper/data
echo 1\2\3 > myid
cd ../bin
./zkServer.sh start
./zkServer.sh status

server.1=192.168.233.61:3188:3288
server.2=192.168.233.62:3188:3288
server.3=192.168.233.63:3188:3288

1: 指的是服务器的序号

192.168.233.61: 对应序号的服务器的ip地址

3188: leader和follower之间交换信息的端口

3288: 选举端口

kafka

kafka消息对列：服务端向客户端发送一个指令，客户端收到指令并且通过这个指令反馈到服务器，完整一个异步方式的通信的流程

kafka消息队列：大数据场景非常合适

rabbitMQ：适合处理小场景的数据请求

activeMQ：适合处理小场景的数据请求

消息队列的应用场景：

1、异步处理

用户的短信验证码，邮件通知

2、系统解耦

微服务架构中的服务之间通信

降低各个组件之间的依赖程度（耦合度），可以提高组件的灵活性和可维护性

3、负载均衡

高并发系统中的任务处理，消息对列把所有任务根发到多个消费者实列，电商平台的订单系统

4、流量控制和限流

根据api请求来进行处理，通过控制消息的生产速度和消费的处理速度来完成限流

端口：应用和应用之间通信

api接口：应用程序内部各个组件之间通信的方式

5、任务调度和定时任务

消息队列可以定时的进行任务调度，按照消费者的计划生成对应的数据

6、数据同步和分发：

日志收集和数据收集，远程的实现数据的统一管理

7、实时数据处理

8、备份和恢复

消息队列的模式：
1、点对点，一对一，生产者生产消息，消费者消费信息，这个是一对一的

2、发布/订阅模式

消息的生产者发布一个主题，其他的消费者订阅这个主题，从而实现一对多

主题：topic

kafka组件的名称：

1、主题 topic，主题是kafka数据流的一个基本的单元，类似于数据的管道。生产者将消息发布到主题，其他的消费者订阅主题，来消费消息，主题可以被分区，分区有偏移量

2、生产者: 将消息写入主题和分区

3、消费者：从分区和主题当中消费发布的消息,一个消费者可以订阅多个主题

4、分区：一个主题可以被分为多个分区，每个分区都是数据的有序子集，分区越多，消费者消费的速度就快，避免生产者的消费堆积，分区当中有偏移量，按照偏移量进行有序存储，消费者可以独立的读写每个分区的数据，如何读取分区的数据，一般是不考虑的，只有在消息出现堆积的时候，会考虑扩展分区数

kafka的消费堆积出现应该如何处理：消费者没有及时处理掉生产者发布的消息，导致消息堆积，可以扩展分区数

5、偏移量：消息在分区当中的唯一标识,根据偏移量指定消费者获取消息的位置

6、经纪人：zookeeper，存储分区的信息，kafka集群的元数据

cd /opt
把kafka拖过去
free -h
echo 3 > /proc/sys/vm/drop_caches
tar -xf  kafka
mv kafka  /usr/local/kafka
cd /usr/local/kafka/config
cp server.properties server.properties.bak
vim server.properties
第21行
id号要不一样
第34行去掉注释
listeners=PLAINTEXT：//:192.168.42.20:9092
第65行
log.dirs=/usr/local/kafka/logs
第103行
log.retention.hours=168
#消息队列当中，数据保存的最大时间，默认是7天，单位是小时
第130行
zookeeper.connect=192.168.42.20:2181,192.168.42.30:2181,192.168.42.40:2181
保存退出
cd /usr/local/kafka/bin
./kafka-server-start.sh -daemon /usr/local/kafka/config/server.properties
netstat -antp | grep 9092

nginx1

vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
保存退出
source /etc/profile
cd /usr/local/kafka/bin
kafka-topics.sh --create --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42,40:9092 --replication-factor 2 --partitions 3 --topic test1
cd /usr/local/kafka/bin
kafka-console-producer.sh --broker-list 192.168.42.20:9092,192.168.42.30:9092,192.168.42,40:9092  --topic test1
kafka-console-producer.sh --broker-list 192.168.42.20:9092,192.168.42.30:9092,192.168.42.40:9092 --topic test1

nginx2

cd /usr/local/kafka/bin
./kafka-topics.sh --list --bootstrap-server 192.168.42.20:9092

nginx3

vim /etc/profile
export KAFKA_HOME=/usr/local/kafka
export PATH=$PATH:$KAFKA_HOME/bin
保存退出
source /etc/profile
cd /usr/local/kafka/bin
kafka-console-consumer.sh --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42.40:9092  --topic test1 --from-beginning

kafka-topics.sh --create --bootstrap-server 192.168.233.61:9092,192.168.233.62:9092,192.168.233.63:9092 --replication-factor 2 --partitions 3 --topic test1

bootstrap-server: 这里的地址一般是一个集群当中地址即可，默认的我们包含所有。
–replication-factor 2: 定义主题的副本数2个副本，2-4个副本是偶数。
partitions 3: 定义分区数 3 6个

修改主题的分区数： 
cd /opt
查看
kafka-topics.sh --describe --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42,40:9092 | grep test1
修改
kafka-topics.sh --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42,40:9092 --alter --topic test1 --partitions 6
删除主题
kafka-topics.sh --delete --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42,40:9092 --topic test1

mysql1 
cd /usr/local/filebeat
vim filebeat.yml
第177行注释掉
output.kafka: enables: truehosts: ["192.168.42.20:9092","192.168.42.30:9092","192.168.42.40:9092"]topic: nginx_mysql
保存退出
./filebeat -e -c filebeat.ymlelk-lk-3
cd /etc/logstash/conf.d
vim kafka.conf
input {kafka {bootstrap_servers => "192.168.42.20:9092,"topics => "nginx_mysql"type => "nginx_kafka"codec => "json"#指定数据的格式jsonauto_offset_reset => "latest"#latest，从尾部开始decorate_events => true#传递给饿死的数据额外的附加的kafka数据}
}
output {if "nginx-1" in [tags] {elasticsearch {hosts => ["192.168.42.60:9200", "192.168.42.61:9200"]index => "%{[fields][serrvice_name]}-%{+YYYY.MM.dd}"}}if "mysql-1" in [tags] {elasticsearch {hosts => ["192.168.42.60:9200", "192.168.42.61:9200"]index => "%{[fields][serrvice_name]}-%{+YYYY.MM.dd}"}}
}
保存退出
logstash -f kafka.conf --path.data /opt/testa &nginx1
kafka-console-consumer.sh --bootstrap-server 192.168.42.20:9092,192.168.42.30:9092,192.168.42.40:9092  --topic nginx_mysql --from-beginning

ELK