架构简介EElasticsearch、LLogstash、FFIlebeat、KKibana业务服务器日志 → Filebeat采集→ Logstash处理→ Elasticsearch存储/检索→ Kibana可视化1. Filebeat日志采集定位轻量级日志采集工具Beats 家族成员部署在日志产生的服务器上。核心作用实时监控指定日志文件如应用日志、系统日志、Nginx 日志等支持日志轮转、多文件匹配。轻量低耗资源占用远低于 Logstash适合在生产服务器批量部署。将采集到的日志初步整理后发送到 Logstash用于复杂处理或直接发送到 Elasticsearch简单场景。2. Logstash日志处理定位日志过滤与转换引擎是 ELFK 中的 “数据加工厂”。核心作用接收 Filebeat 发送的原始日志进行结构化处理如解析非 JSON 日志、提取关键字段。支持数据清洗过滤无用字段、脱敏敏感信息、格式转换如统一时间戳格式、数据 enrichment关联外部数据。将处理后的结构化日志转发到 Elasticsearch 存储。3. Elasticsearch日志存储与检索定位分布式全文搜索引擎负责日志的存储、索引与快速检索。核心作用以 JSON 格式存储结构化日志通过分片和副本机制实现高可用与高吞吐。基于倒排索引支持秒级全文检索可按关键词、时间范围、字段条件快速筛选日志。支持水平扩展通过增加节点提升存储容量和检索性能。4. Kibana日志可视化定位Elasticsearch 的可视化前端工具提供日志分析与展示界面。核心作用通过 Web 界面连接 Elasticsearch支持日志实时查询、筛选、导出。提供丰富的可视化组件如折线图、柱状图、饼图、仪表盘可自定义监控面板如系统错误率、接口访问量趋势。支持创建告警规则当日志中出现异常模式如错误日志激增时触发告警。在 ELFK 架构中加入 Kafka 后通过 Kafka 作为日志传输的中间缓冲层解决高并发场景下日志峰值冲击、组件解耦、异步处理等问题使架构更稳定、可扩展。Filebeat采集→ Kafka缓冲→ Logstash处理→ Elasticsearch存储→ Kibana可视化加入 Kafka 的核心优势削峰填谷应对日志峰值当业务突发流量导致日志量激增如秒杀活动、系统故障时的错误日志爆发Kafka 可暂存大量日志避免直接冲击 Logstash 或 Elasticsearch 导致组件过载。Logstash 可按自身处理能力从 Kafka 消费保证下游组件稳定。解耦组件提升架构灵活性Filebeat 仅需关注日志采集并发送到 Kafka无需关心后续处理组件Logstash 可独立升级或替换。除 Logstash 外其他系统如 Flink 实时计算、数据仓库可同时从 Kafka 消费日志实现日志的多目的地分发一份日志供多个场景使用。异步处理提高系统吞吐量日志流转从 “同步链路”Filebeat → Logstash → ES变为 “异步缓冲”Filebeat → Kafka 异步写入Logstash 异步消费减少组件间的直接依赖提升整体吞吐量。日志可靠性保障Kafka 支持消息持久化和多副本机制replication-factor ≥ 2即使 Logstash 或 ES 短暂故障日志也不会丢失待下游恢复后可继续消费。集群部署环境配置新建三台虚拟机做好静态IP解析关闭防火墙三台节点sever678都安装JDK环境验证再下载kafka压缩包解压备份配置文件创建目录文件三台主机 Kafka 配置文件修改修改配置文件server.properties参数详解# 节点角色。Kafka节点同时作为Broker和Controller运行。Controller负责集群管理比如分区分配和Leader选举而Broker处理消息的存储和转发。process.rolesbroker,controller#节点的唯一标识符每个Broker都需要一个不同的ID确保在集群中唯一。node.id157#Controller的仲裁节点配置了三个节点分别是157、158和159。在选举Controller Leader时需要这三个节点中的多数同意确保高可用性。controller.quorum.voters157192.168.36.157:9093,158192.168.36.158:9093,159192.168.36.159:9093# 定义了Broker监听地址和端口。两个监听器PLAINTEXT和CONTROLLER分别对应不同的端口。PLAINTEXT用于普通客户端通信CONTROLLER用于Controller之间的内部通信。listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093# 指定了Broker之间内部通信使用的监听器名称这里是PLAINTEXT意味着Broker之间使用明文协议通信。inter.broker.listener.namePLAINTEXT#客户端连接Broker时使用的地址同样配置了PLAINTEXT和CONTROLLER但客户端通常使用PLAINTEXT来连接。advertised.listenersPLAINTEXT://192.168.36.157:9092,CONTROLLER://192.168.36.157:9093# Controller使用CONTROLLER监听器进行内部通信确保Controller之间的消息传输。controller.listener.namesCONTROLLERlistener.security.protocol.mapCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL# 处理网络请求的最大线程数num.network.threads32# 处理I/O请求的线程数num.io.threads32# TCP发送和接收缓冲区的大小通常设置为102400字节。socket.send.buffer.bytes102400socket.receive.buffer.bytes102400# 限制了单个请求的最大大小防止内存溢出此处为100MB。socket.request.max.bytes104857600# 指定日志存储的目录log.dirs/opt/kafka/logs# 默认的分区数新建Topic时如果没有指定会使用这个值影响并行处理能力。num.partitions3# 控制日志恢复时的线程数每个数据目录一个线程加快恢复速度。num.recovery.threads.per.data.dir1# 确保偏移量和事务日志的高可用性副本数设为3容忍两个节点故障。offsets.topic.replication.factor3transaction.state.log.replication.factor3# 设置事务日志的最小同步副本数保证至少有一个副本同步完成。transaction.state.log.min.isr1# 允许删除Topic但需谨慎使用。delete.topic.enabletrue# 控制日志保留策略按时间和大小删除旧数据。log.retention.hours168log.retention.bytes1073741824# 涉及生产者和消费者的性能调优比如批量消息大小和内存缓冲区。buffer.memory68719476736batch.size2097152max.request.size4194304message.max.bytes6291456fetch.max.bytes7340032linger.ms10compression.typesnappylog.segment.bytes1073741824log.retention.check.interval.ms300000修改配置文件由于要修改的较多这里直接展示模板三个节点都可以按照这个来修改process.rolesbroker,controllernode.id172controller.quorum.voters172192.168.154.172:9093,158192.168.154.173:9093,159192.168.154.174:9093 #三个节点IP的9093端口listenersPLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093inter.broker.listener.namePLAINTEXTadvertised.listenersPLAINTEXT://192.168.154.172:9092,CONTROLLER://192.168.154.172:9093controller.listener.namesCONTROLLERlistener.security.protocol.mapCONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSLnum.network.threads2num.io.threads2socket.send.buffer.bytes102400socket.receive.buffer.bytes102400socket.request.max.bytes104857600log.dirs/opt/kafka/logsnum.partitions3num.recovery.threads.per.data.dir1offsets.topic.replication.factor3transaction.state.log.replication.factor3transaction.state.log.min.isr1log.retention.hours168log.segment.bytes1073741824log.retention.check.interval.ms300000笔者这里直接将文件拷贝之后再去sever78修改节点号和ip即可Kafka 集群初始化server7 节点生成储目录唯一的 UUID三台主机用该 uuid 格式化 kafka 存储目录笔者这里发现配置文件有错误修改了一下所以上下文的uuid出现不同三台主机启动kafkaKafka 集群可用性验证Kafka 集群任意节点创建 Topic--replication-factor 3设置主题的副本因子为 3即每个分区的副本数量为 3。副本用于数据冗余和高可用性副本数需不能超过集群中 Broker 的总数--partitions 3设置主题的分区数为 3。分区是 Kafka 并行处理消息的基本单位影响吞吐量和扩展性。分区数一旦创建不可减少Kafka 集群任意节点查看 TopicKafka 集群任意节点生产者测试Kafka 集群任意节点查看主题Kafka 集群任意节点查看 topic 详细信息Filebeat 配置文件修改sever6安装filebeat添加kafka输出参数这里注意缩进上面的output记得注释掉测试语法测试连接重启服务Kafka 集群任意节点消费者测试Logstash 消费 Topic 消息sever5新建文件添加内容读取文件成功