标签：ELK Filebeat -- zookeeper 192.168 kafka 2181 Kafka

首先介绍一下什么是Filebeat+Kafka+ELK，ELK+Filebeat就是通过Filebeat收集数据，然后通过logstash发送到es上，然后直接发送的话，如果数据过大，或者遇到其他别的一些问题，在高并发环境下，同步请求来不及处理，请求往往会发生阻塞。所以我们使用Kafka通过异步处理请求，从而缓解系统的压力。Kafka就是消息队列，把Filebeat收集的数据输出到Kafka，然后通过logstash把经过消息队列排序的消息有序的发往es储存，从而避免一个时间系统请求过多带来的系统崩盘。

一、环境准备

node1节点（内存4G）：192.168.235.101 部署ES、Kibana、Filebeat
node2节点（内存4G）：192.168.235.102 部署ES
apache节点：192.168.235.103 部署logstash、apache
zookeeper集群(部署zookeeper)：
server1：192.168.235.104
server2：192.168.235.105
server3：192.168.235.106
关闭防火墙
以下实验建立在以上环境当中，已经全部完成，详情见之前博客。

二、安装kafka

在zookeeper集群中三台机子上均要安装


修改配置文件






修改环境变量



配置启动脚本

#!/bin/bash
#chkconfig:2345 22 88
#description:Kafka Service Control Script
KAFKA_HOME='/usr/local/kafka'
case $1 in
start)
	echo "---------- Kafka 启动 ------------"
	${KAFKA_HOME}/bin/kafka-server-start.sh -daemon ${KAFKA_HOME}/config/server.properties
;;
stop)
	echo "---------- Kafka 停止 ------------"
	${KAFKA_HOME}/bin/kafka-server-stop.sh
;;
restart)
	$0 stop
	$0 start
;;
status)
	echo "---------- Kafka 状态 ------------"
	count=$(ps -ef | grep kafka | egrep -cv "grep|$$")
	if [ "$count" -eq 0 ];then
        echo "kafka is not running"
    else
        echo "kafka is running"
    fi
;;
*)
    echo "Usage: $0 {start|stop|restart|status}"
esac

创建topic

查看当前服务器中的所有 topic
kafka-topics.sh --list --zookeeper 192.168.235.104:2181,192.168.235.105:2181,192.168.235.106:2181
查看某个 topic 的详情
kafka-topics.sh  --describe --zookeeper 192.168.235.104:2181,192.168.235.105:2181,192.168.235.106:2181
发布消息
kafka-console-producer.sh --broker-list 192.168.235.104:9092,192.168.235.105:9092,192.168.235.106:9092  --topic name1
消费消息
kafka-console-consumer.sh --bootstrap-server 192.168.235.104:9092,192.168.235.105:9092,192.168.235.106:9092 --topic name1 --from-beginning
--from-beginning：会把主题中以往所有的数据都读取出来
修改分区数
kafka-topics.sh --zookeeper 192.168.235.104:2181,192.168.235.105:2181,192.168.235.106:2181 --alter --topic name1 --partitions 6
删除 topic
kafka-topics.sh --delete --zookeeper 192.168.235.104:2181,192.168.235.105:2181,192.168.235.106:2181 --topic name1

三、Filebeat+Kafka+ELK

修改Filebeat配置文件



启动 filebeat

在logstash上新建一个配置文件

input {
    kafka {
        bootstrap_servers => "192.168.235.104:9092,192.168.235.105:9092,192.168.235.106:9092"
        topics  => "filebeat_name1"
        group_id => "name123"
        auto_offset_reset => "earliest"
    }
}

output {
    elasticsearch {
        hosts => ["192.168.235.101:9200"]
        index => "filebeat-%{+YYYY.MM.dd}"
    }
    stdout {
        codec => rubydebug
    }
}

启动 logstash

登录 Kibana


可以查看到node1节点的信息，以上的操作是使用filebeat收集了node1的数据，传送到zookeeper集群，zookeeper这边是通过kafka接受，kafka作为一个消息队列，他可以对数据进行处理，然后传送给通过logstash传送到es中进行储存，最后通过kibana展示出来。

标签：ELK,Filebeat,--,zookeeper,192.168,kafka,2181,Kafka
来源： https://blog.csdn.net/weixin_55609951/article/details/118978299

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。