前面提到过，logstash占资源很大，filebeat更加轻量，一般都是组合使用，难免会有logstash宕掉的时候，这时候filebeat再往logstash里写数据就写不了了，这期间的日志信息可能就无法采集到了，因此一般都会采用redis或kafka作为一个消息缓冲层，本篇就介绍如何使用Redis作为数据缓冲层，将filebeat的数据落地到Redis中然后由Logstash消费并写入至ES。
ELK系列(一)、安装ElasticSearch+Logstash+Kibana+Filebeat-v7.7.0
ELK系列(二)、在Kibana中使用RESTful操作ES库
ELK系列(四)、Logstash读取nginx日志写入ES中
ELK系列(五)、Logstash修改@timestamp时间为日志的产生时间
ELK系列(六)、修改Nginx日志为Json格式并使用Logstash导入至ES
ELK系列(七)、Filebeat+Logstash采集多个日志文件并写入不同的ES索引中
-------------------------------使用Filebeat+Redis+Logstash收集日志数据------------------------------
首先配置Filebeat的yml文件，这里还使用之前用过的采集nginx的access-json.json日志的配置文件改一改，采集nginx的访问日志并写入到redis中，指定key为"nginx",在db0内，数据类型为list：
vim /opt/app/filebeat-7.7.0-linux-x86_64/filebeat_redis.yml
filebeat.inputs:
- type: log
enabled: true
backoff: "1s"
tail_files: false
paths:
- /usr/local/nginx/logs/access-json.log
fields:
filetype: log_nginxjson
fields_under_root: true
output.redis:
enabled: true
hosts: ["wykd:6379"]
password: 123456
key: nginx
db: 0
datatype: list如果redis没有开启密码的话，这里的password参数可以不写。更加详细的参数配置可参考$FILEBEAT_HOME下的filebeat.reference.yml文件。
logstash的配置文件也拿之前的读nginx日志文件写到es的配置文件改一改，把input改成从redis取数据，这里的password也是按需配置，key,data_type,db,host,port需要与上面filebeat配置的一致。
vim /opt/app/logstash-7.7.0/config/logstash_nginxlog2es_redis.conf
input {
# 从文件读取日志信息
redis {
host => "wykd"
port => 6379
password => "123456" #如果没有密码不需要写这个参数
key => "nginx"
data_type => "list"
db =>0
}
}
filter {
json {
source => "message"
remove_field => ["beat","offset","tags","prospector"] #移除字段，不需要采集
}
date {
match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"] #匹配timestamp字段
target => "@timestamp" #将匹配到的数据写到@timestamp字段中
}
}
output {
# 输出es
elasticsearch {
hosts => ["wykd:9200"]
index => "redis-%{+YYYY.MM.dd}"
}
}首先启动logstash，然后启动filebeat：
cd $LS_HOME
bin/logstash -f config/logstash_nginxlog2es_redis.conf
cd $FILEBEAT_HOME
./filebeat -e -c filebeat_redis.yml进入kibana查看index，可以看到redis-*的index已经创建出来了，并且有14条数据:
访问nginx网页三次进行测试:
curl http://wykd/wyk.html
curl http://wykd/abc.html
curl http://wykd可以看到这三次访问记录已经进来了，变成了17条记录：
此时我们关闭logstash进程，模拟一下logstash关闭的情况下，如果继续给nginx新增日志的话，filebeat和redis里的数据是什么样的：
1. 关闭logstash
2. 模拟登陆nginx:
curl http://wykd/111.html
curl http://wykd/222.html
3. 查看redis数据
可以看到在redis的db0的key=nginx里已经缓存了这两条日志记录， 当我们再次重启logstash的时候，它就会自动从redis中将缓存的日志记录消费到ES中：
1. 再次启动logstash进程
bin/logstash -f config/logstash_nginxlog2es_redis.conf
2. 验证redis数据
3. 验证es中的index记录数 
线上都是在各应用客户端使用filebeat采集日志，然后推送到redis或kafka集群内，然后由logstash进行消费，此架构可大大提升稳定性。
希望本文对你有帮助，请点个赞鼓励一下作者吧~ 谢谢！