ELK和kafka构建高并发分布式日志收集系统

PaperDragon2023年8月13日...大约 15 分钟

ELK和kafka构建高并发分布式日志收集系统

ELK+Kafka集群

前言

业务层可以直接写入到kafka队列中，不用担心elasticsearch的写入效率问题。图示

Kafka

Apache kafka是消息中间件的一种，是一种分布式的，基于发布/订阅的消息系统。能实现一个为处理实时数据提供一个统一、高吞吐、低延迟的平台，且拥有分布式的，可划分的，冗余备份的持久性的日志服务等特点。

术语

1、kafka是一个消息队列服务器。kafka服务称为broker（中间人）, 消息发送者称为producer（生产者）, 消息接收者称为consumer（消费者）;通常我们部署多个broker以提供高可用性的消息服务集群.典型的是3个broker;消息以topic的形式发送到broker,消费者订阅topic,实现按需取用的消费模式;创建topic需要指定replication-factor(复制数目, 通常=broker数目);每个topic可能有多个分区(partition), 每个分区的消息内容不会重复

2、kafka-broker-中间人

3、webserver/logstash-producer[prəˈdu:sə®]-消息生产者/消息发送者
Producer：
kafka集群中的任何一个broker都可以向producer提供metadata信息,这些metadata中包含"集群中存活的servers列表"/“partitions leader列表"等信息；
当producer获取到metadata信息之后, producer将会和Topic下所有partition leader保持socket连接；
消息由producer直接通过socket发送到broker，中间不会经过任何"路由层”，事实上，消息被路由到哪个partition上由producer客户端决定；比如可以采用"random"“key-hash”"轮询"等,如果一个topic中有多个partitions,那么在producer端实现"消息均衡分发"是必要的。
在producer端的配置文件中,开发者可以指定partition路由的方式。
Producer消息发送的应答机制设置发送数据是否需要服务端的反馈,有三个值0,1,-1
0:producer不会等待broker发送ack
1:当leader接收到消息之后发送ack
-1:当所有的follower都同步消息成功后发送ack

4、elasticsearch-consumer-消费者

5、logs-topic-话题

6、replication-facter-复制数目-中间人存储消息的副本数=broker数目

7、一个topic有多个分区partition
partition：
（1）、Partition：为了实现扩展性，一个非常大的topic可以分布到多个broker（即服务器）上，一个topic可以分为多个partition，每个partition是一个有序的队列。partition中的每条消息都会被分配一个有序的id（offset）。kafka只保证按一个partition中的顺序将消息发给consumer，不保证一个topic的整体（多个partition间）的顺序。
（2）、在kafka中,一个partition中的消息只会被group中的一个consumer消费(同一时刻)；一个Topic中的每个partions，只会被一个consumer消费，不过一个consumer可以同时消费多个partitions中的消息。

实战

说明

说明
1、使用一台Nginx代理访问kibana的请求;
2、两台es组成es集群，并且在两台es上面都安装kibana;（ 以下对elasticsearch简称es ）
3、中间三台服务器就是我的kafka(zookeeper)集群啦; 上面写的 消费者/生产者 这是kafka(zookeeper)中的概念;
4、最后面的就是一大堆的生产服务器啦，上面使用的是logstash，
当然除了logstash也可以使用其他的工具来收集你的应用程序的日志，例如：Flume，Scribe，Rsyslog，Scripts……

角色

1、nginx-proxy(略)：172.16.100.25
2、es1：172.16.100.21
3、es2：172.16.100.24
4、kafka1：172.16.100.26
5、kafka2：172.16.100.27
6、kafka3：172.16.100.32
7、webserver：172.16.100.33

软件说明

1、elasticsearch - 1.7.3.tar.gz
2、Logstash - 2.0.0.tar.gz
3、kibana - 4.1.2 - linux - x64 . tar . gz(略)：
以上软件都可以从官网下载 : https : //www.elastic.co/downloads
4、java - 1.8.0 ， nginx 采用 yum 安装

步骤

1、ES集群安装配置;
2、Logstash客户端配置(直接写入数据到ES集群，写入系统messages日志);
3、Kafka(zookeeper)集群配置;(Logstash写入数据到Kafka消息系统);
4、Kibana部署;
5、Nginx负载均衡Kibana请求;

演示

1、ES集群安装配置

es1：（1）、安装java-1.8.0以及依赖包（每台服务器都安装JAVA）

# yum -y install epel-release
# yum -y install java-1.8.0 git wget lrzsz
#缓存这个java的包
#可以使用只下载不安装，缓存这些包
# yum -y install java-1.8.0 git wget lrzsz --downloadonly --downloaddir=./
#注释： --downloadonly 只下载不安装  downloaddir  目录

（2）、获取es软件包

# wget https://download.elastic.co/elasticsearch/elasticsearch/elasticsearch-1.7.3.tar.gz
# tar -xf elasticsearch-1.7.3.tar.gz -C /usr/local/
# ln -sv /usr/local/elasticsearch-1.7.3/ /usr/local/elasticsearch

（3）、修改配置文件

# vim /usr/local/elasticsearch/config/elasticsearch.yml

cluster.name: es-cluster                         #组播的名称地址
node.name: "es-node1"                            #节点名称，不能和其他节点重复
node.master: true                                #节点能否被选举为master
node.data: true                                  #节点是否存储数据
index.number_of_shards: 5                        #索引分片的个数
index.number_of_replicas: 1                  	 #分片的副本个数
path.conf: /usr/local/elasticsearch/config       #配置文件的路径
path.data: /data/es/data                         #数据目录路径
path.work: /data/es/worker                       #工作目录路径
path.logs: /usr/local/elasticsearch/logs         #日志文件路径
path.plugins: /data/es/plugins                   #插件路径
bootstrap.mlockall: true        				 #内存不向swap交换
http.enabled: true                               #启用http

（4）、创建相关目录

# mkdir -p /data/es/{data,worker,plugins}
#注释：data：放数据的文件 worker：工作临时文件 plugins：插件 日志文件会自己建立

（5）、获取es服务管理脚本

#为了方便配置文件
# git clone https://github.com/elastic/elasticsearch-servicewrapper.git

# mv elasticsearch-servicewrapper/service/ /usr/local/elasticsearch/bin/

#在/etc/init.d/目录下，自动安装上es的管理脚本啦
# /usr/local/elasticsearch/bin/service/elasticsearch install

（6）、启动es，并检查服务是否正常

# systemctl start elasticsearch
# systemctl enable elasticsearch
# ss -nptl |grep -E "9200|9300"
LISTEN     0      50          :::9200                    :::*                   users:(("java",pid=10020,fd=104))
LISTEN     0      50          :::9300                    :::*                   users:(("java",pid=10020,fd=66))

访问192.168.88.153:9200

（3）、安装es的管理插件（1）、说明：es官方提供一个用于管理es的插件，可清晰直观看到es集群的状态，以及对集群的操作管理，安装方法如下：（2）、提示：

# /usr/local/elasticsearch/bin/plugin -i mobz/elasticsearch-head

安装好之后，访问方式为： http://192.168.0.110:9200/_plugin/head，由于集群中现在暂时没有数据，所以显示为空,

2、Logstash客户端安装配置（在webserver1上安装logstassh，用于采集日志）

（1）、downloads 软件包

# yum -y install java-1.8.0
# wget https://download.elastic.co/logstash/logstash/logstash-2.0.0.tar.gz
# tar -xf logstash-2.0.0.tar.gz -C /usr/local
# cd /usr/local/
# ln -sv logstash-2.0.0 logstash
# mkdir /usr/local/logstash/{logs,etc}

（2）、Logstash 向es集群写数据编写配置文件;

# vim /usr/local/logstash/etc/logstash.conf
input {                #数据的输入从标准输入
stdin { }
}
output {              #数据的输出我们指向了es集群
elasticsearch {
hosts => [ "192.168.0.110:9200" , "192.168.0.111:9200" ] #es 主机的 ip 及端口
}
}

检查配置文件是否有语法错误：

#  /usr/local/logstash/bin/logstash -f /usr/local/logstash/etc/logstash.conf --configtest --verbose

输出提示：
Configuration OK		语法正确

（3）、启动logstash 启动;

# /usr/local/logstash/bin/logstash -f /usr/local/logstash/etc/logstash.conf

4、扩展（Nginx负载均衡Kibana的请求）

（1）、在nginx-proxy上面yum安装nginx

# yum install -y nignx

（2）、编写配置问佳佳es.conf

# vim /etc/nginx/conf.d/es.conf
upstream es {
server 172.16.100.21:5601 max_fails=3 fail_timeout=30s;
server 172.16.100.24:5601 max_fails=3 fail_timeout=30s;
}
 
server {
    listen       80;
    server_name  localhost;
 
    location / {
        proxy_pass http://es/;
        index index.html index.htm;
        #auth
        auth_basic "ELK Private";
        auth_basic_user_file /etc/nginx/.htpasswd;
    } 
}

（3）、创建认证

3.创建认证
# htpasswd -cm /etc/nginx/.htpasswd elk
New password:
Re-type new password:
Adding password for user elk-user
# /etc/init.d/nginx restart
Stopping nginx:                                            [  OK  ]
Starting nginx:                                            [  OK  ]

（4）、访问

1、解耦

允许你独立的扩展或修改两边的处理过程，只要确保它们遵守同样的接口约束

2、冗余

消息队列把数据进行持久化直到它们已经被完全处理，通过这一方式规避了数据丢失风险。许多消息队列所采用的"插入-获取-删除"范式中，在把一个消息从队列中删除之前，需要你的处理系统明确的指出该消息已经被处理完毕，从而确保你的数据被安全的保存直到你使用完毕。

3、扩展性

因为消息队列解耦了你的处理过程，所以增大消息入队和处理的频率是很容易的，只要另外增加处理过程即可。

4、灵活性 & 峰值处理能力

在访问量剧增的情况下，应用仍然需要继续发挥作用，但是这样的突发流量并不常见。如果为以能处理这类峰值访问为标准来投入资源随时待命无疑是巨大的浪费。使用消息队列能够使关键组件顶住突发的访问压力，而不会因为突发的超负荷的请求而完全崩溃。

5、可恢复性

系统的一部分组件失效时，不会影响到整个系统。消息队列降低了进程间的耦合度，所以即使一个处理消息的进程挂掉，加入队列中的消息仍然可以在系统恢复后被处理。

6、顺序保证

在大多使用场景下，数据处理的顺序都很重要。

大部分消息队列本来就是排序的，并且能保证数据会按照特定的顺序来处理。（Kafka 保证一个 Partition 内的消息的有序性） 7、缓冲

有助于控制和优化数据流经过系统的速度，解决生产消息和消费消息的处理速度不一致的情况。

8、异步通信

很多时候，用户不想也不需要立即处理消息。消息队列提供了异步处理机制，允许用户把一个消息放入队列，但并不立即处理它。想向队列中放入多少消息就放多少，然后在需要的时候再去处理它们。

Redis与Kafka

我们都知道Redis是以key的hash方式来分散对列存储数据的，且Redis作为集群使用时，对应的应用对应一个Redis，在某种程度上会造成数据的倾斜性，从而导致数据的丢失。

而从之前我们部署Kafka集群来看，kafka的一个topic（主题），可以有多个partition（副本），而且是均匀的分布在Kafka集群上，这就不会出现redis那样的数据倾斜性。Kafka同时也具备Redis的冗余机制，像Redis集群如果有一台机器宕掉是很有可能造成数据丢失，而Kafka因为是均匀的分布在集群主机上，即使宕掉一台机器，是不会影响使用。同时Kafka作为一个订阅消息系统，还具备每秒百万级别的高吞吐量，持久性的、分布式的特点等。