加入kafka缓存

kafka和zookeeper参考消息队列里面的安装
nginx服务器安装logstash
[root@za-proxy-active ~]# cd /etc/logstash/conf.d/
[root@za-proxy-active conf.d]# cat m43-log-to-kafka.conf
input {
file {
path => "/var/log/messages"
type => "systemlog"
start_position => "beginning"
stat_interval => "3"
}

file {
path => "/apps/nginx/logs/access.log"
type => "nginx-acceslog"
start_position => "beginning"
stat_interval => "3"
codec => json
}

file {
path => "/apps/nginx/logs/error.log"
type => "nginx-errorlog"
start_position => "beginning"
stat_interval => "3"
}
}

output {
if [type] == "systemlog" {
kafka {
bootstrap_servers => "10.0.0.111:9092,10.0.0.112:9092,10.0.0.113:9092"
topic_id => "m43-systemlog"
codec => "json"
}}

if [type] == "nginx-acceslog" {
kafka {
bootstrap_servers => "10.0.0.111:9092,10.0.0.112:9092,10.0.0.113:9092"
topic_id => "nginx-acceslog"
codec => "json"
}}

if [type] == "nginx-errorlog" {
kafka {
bootstrap_servers => "10.0.0.111:9092,10.0.0.112:9092,10.0.0.113:9092"
topic_id => "nginx-errorlog"
codec => "json"
}}

stdout {}

}
重启服务
systemctl restart logstash.service
检查语法
[root@za-proxy-active conf.d]# /usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/m43-redis-to-es.conf -t
写入数据或者访问nginx产生日志
echo "123" >> /var/log/messages

去kafka机器查看是否加载
root@mq-server1:~# kafka-topics.sh --list --zookeeper 10.0.0.111:2181
m43-systemlog
m44
nginx-acceslog

logstash服务器，读取kafka服务器数据，传送给es集群
cd /etc/logstash/conf.d/
mv ./* /opt/
root@zhaohuakang:conf.d# cat m43-kafka-to-es.conf
input {
kafka {
bootstrap_servers => "10.0.0.111:9092,10.0.0.112:9092,10.0.0.113:9092"
topics => ["m43-systemlog","nginx-errorlog","nginx-acceslog"]
codec => "json"
}

}

output {
if [type] == "nginx-acceslog" {
elasticsearch {
hosts => ["10.0.0.101:9200"]
index => "logstash-m43-nginx-accesslog-%{+YYYY.MM.dd}"
}}

if [type] == "nginx-errorlog" {
elasticsearch {
hosts => ["10.0.0.101:9200"]
index => "logstash-m43-nginx-errorlog-%{+YYYY.MM.dd}"
}}

if [type] == "systemlog" {
elasticsearch {
hosts => ["10.0.0.101:9200"]
index => "logstash-m43-systemlog-%{+YYYY.MM.dd}"
}}
}

测试语法
/usr/share/logstash/bin/logstash -f /etc/logstash/conf.d/m43-kafka-to-es.conf -t
重启服务
systemctl restart logstash.service

kibana展示