基于Kafka（Kraft）+ELK的日志收集分析平台搭建（上）

1. 完整业务链路梳理本项目搭建一套企业标准全链路日志处理平台实现流量接入→日志采集→消息缓冲→多分支消费处理完整闭环整体数据流用户浏览器访问域名 www.sc.com三台 Nginx 反向代理集群做流量分发转发至后端 Flask 业务服务每台 Nginx 本地生成 access/error 访问日志本机 Filebeat 采集日志三节点 KRaft 模式 Kafka 集群作为消息中间件完成流量削峰、解耦Kafka 下游三条并行消费分支日志清洗服务结构化清洗日志存入 ElasticsearchKibana 绘制访问监控看板告警检测服务自定义错误阈值规则触发后推送 QQ / 邮件告警通知运维AI 智能分析服务结合 LangChain 本地大模型自动解析报错日志、输出故障排查方案配套中间件Redis 缓存告警规则、统计数据MySQL 持久化告警记录与报表数据详细流程客户端访问 → Nginx 反向代理集群三台 Flask 业务服务 → 每台机器 Filebeat 采集access.log/error.log→ KRaft 模式 Kafka 集群缓冲削峰 → 下游多消费端并行处理分支 1日志清洗 Agent → Elasticsearch 存储 Kibana 可视化看板分支 2告警分析 Agent → 邮件 / QQ 告警推送分支 3对接 LangChain 本地部署大模型做日志智能解析、故障研判配套中间件Redis 缓存、MySQL 持久化分析结果2. 技术栈完整清单基础运维Linux中间件Kafka 3.6.1KRaft 无 ZooKeeper、MySQL采集层Filebeat存储可视化Elasticsearch、Kibana智能拓展LangChain、本地大模型运维能力日志清洗、日志告警、告警规则自定义部署规划表主机名部署服务192.168.140.131nginx Filebeat Kafka MySQL192.168.140.131nginx Filebeat Kafka192.168.140.131nginx Filebeat Kafka Elasticsearch Kibana部署说明Nginx、Filebeat 三台全部部署搭建反向代理高可用集群每台 Nginx 独立产生日志遵循日志采集「本地读取」原则Kafka 三台混合部署 brokercontroller标准 KRaft 集群架构三副本保证日志数据高可用部署 Elasticsearch 与 MySQL将数据持久化存储Kibana 实现日志分析可视化一、集群基础环境统一初始化三台机器全部执行1.1 基础依赖工具安装dnf install wget vim java-21-openjdk.x86_64 net-tools -y1.2 配置静态 IP编辑网卡配置文件/etc/NetworkManager/system-connections/ens33.nmconnection[ipv4] methodmanual addresses1192.168.140.130/24,192.168.140.2 dns114.114.114.114重载网络配置# 重新加载配置 nmcli c reload 激活 ens33 物理网卡让刚重载的静态 IP 配置正式生效 nmcli d up ens331.3 配置主机名 /etc/hosts 域名映射配置主机名以 Kafka1 为例hostnamectl set-hostname kafka1修改/etc/hosts文件添加主机名和 IP 地址映射# 修改 /etc/hosts vim /etc/hosts 添加主机名和 IP 地址映射 192.168.140.130 kafka1 192.168.140.129 kafka2 192.168.140.134 kafka31.4 关闭防火墙与 SELinux# 清空防火墙规则 iptables -F 关闭防火墙开机自启动 systemctl stop firewalld systemctl disable firewalld 永久关闭 SELinux sed -i s/SELINUXenforcing/SELINUXdisabled/ /etc/selinux/config 重启系统生效 reboot二、KRaft 模式 Kafka 集群部署2.1 Kafka 下载、解压缩三台全部执行cd /opt 官方下载安装包 wget https://archive.apache.org/dist/kafka/3.6.1/kafka_2.13-3.6.1.tgz 解压缩 tar xf kafka_2.13-3.6.1.tgz cd kafka_2.13-3.6.12.2 修改核心配置文件位于 kafka 目录 config/kraft/server.propertiesKRaft 架构取消 Zookeeper单节点同时承担 broker 消息存储 controller 元数据管理角色三台仅修改 node.id、监听 IP其余配置保持统一。以 Kafka1 的修改为例# 修改节点 id每个节点唯一 node.id1 修改控制器投票列表 controller.quorum.voters1192.168.140.130:9093,2192.168.140.134:9093,3192.168.140.129:9093 修改监听器和控制器绑定 ip。其中 kafka1 为主机名可用本机 ip 地址代替 listenersPLAINTEXT://kafka1:9092,CONTROLLER://kafka1:9093 侦听器名称、主机名和代理将向客户端公布的端口broker 对外暴露的地址 如果未设置则使用 listeners 的值 advertised.listenersPLAINTEXT://kafka1:90922.3 集群元数据初始化2.3.1 生成全局唯一集群 UUID在 Kafka1任意一台机器上执行cd /opt/kafka_2.13-3.6.1 bin/kafka-storage.sh random-uuid # 输出示例z3oq9M4IQguOBm2rt1ovmQ2.3.2 格式化存储目录三台机器全部执行格式化命令-t 后面接自己生成的 UUIDbin/kafka-storage.sh format -t z3oq9M4IQguOBm2rt1ovmQ -c /opt/kafka_2.13-3.6.1/config/kraft/server.properties2.3.3 启动 Kafka 服务# 启动 bin/kafka-server-start.sh -daemon /opt/kafka_2.13-3.6.1/config/kraft/server.properties 关闭 bin/kafka-server-stop.sh输出成功示例三台机器都要启动成功2.4 systemd 托管 Kafka 服务后台常驻、开机自启Kafka 可以用脚本直接后台启动但集群环境推荐交给 systemd 托管为系统服务。2.4.1 开机自启脚本后台运行不会持久化服务器重启 Kafka 就停止需要手动逐台启动systemd 配置开机自启机器重启自动拉起服务防止节点漏启动。2.4.2 运维命令标准化统一使用systemctl start/stop/restart/status kafka一条命令查看状态与报错不用记忆冗长的脚本路径便于多台机器批量运维。创建服务文件/usr/lib/systemd/system/kafka.service[Unit] DescriptionApache Kafka server (KRaft mode) Documentationhttp://kafka.apache.org/documentation.html Afternetwork.target [Service] Typeforking Userroot Grouproot EnvironmentPATH/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/root/bin:/usr/lib/jvm/java-21-openjdk/bin/ ExecStart/opt/kafka_2.13-3.6.1/bin/kafka-server-start.sh -daemon /opt/kafka_2.13-3.6.1/config/kraft/server.properties ExecStop/opt/kafka_2.13-3.6.1/bin/kafka-server-stop.sh Restarton-failure [Install] WantedBymulti-user.target加载配置并启动集群# 重新加载 systemd 配置 systemctl daemon-reload 启动 kafka 服务 systemctl start kafka 关闭 kafka 服务 systemctl stop kafka 设置开机自启 systemctl enable kafka2.5 测试集群# 创建 topic bin/kafka-topics.sh --create --bootstrap-server kafka3:9092 --replication-factor 3 --partitions 3 --topic my_topic # --replication-factor 指定副本因子--partitions 指定分区数--topic 指定主题名称 查看 topic bin/kafka-topics.sh --list --bootstrap-server kafka3:9092 创建生产者发送消息测试用 bin/kafka-console-producer.sh --broker-list kafka3:9092 --topic my_topic 创建消费者获取数据测试用 bin/kafka-console-consumer.sh --bootstrap-server kafka1:9092 --topic my_topic --from-beginning三、Nginx 反向代理集群部署3.1 Nginx 集群部署三台机器全部安装dnf install nginx -y3.2 反向代理负载均衡配置三台 Nginx 配置完全统一文件路径/etc/nginx/conf.d/sc.confupstream flask { # 仅部署了 Flask 的两台后端服务 IP server 192.168.140.129:5000; server 192.168.140.134:5000; } server { listen 80; server_name www.sc.com; location / { proxy_pass http://flask; # 传递真实客户端 IP日志中可记录用户真实地址 proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; } }3.3 Nginx 配置校验与启停# 校验配置语法 nginx -t 启动并开机自启 systemctl start nginx systemctl enable nginx四、Filebeat 安装部署4.1 下载解压wget https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.19.19-linux-x86_64.tar.gz tar -zxvf filebeat-8.19.19-linux-x86_64.tar.gz -C /usr/local/ mv /usr/local/filebeat-8.19.19-linux-x86_64 /usr/local/filebeat cd /usr/local/filebeat4.2 备份原配置cp filebeat.yml filebeat.yml.bak4.3 修改 filebeat.yml 配置# 采集日志输入 filebeat.inputs: - type: filestream enabled: true paths: - /usr/local/nginx/logs/access.log 关闭 ES 默认输出 output.elasticsearch: enabled: false 输出到 Kafka作为生产者 output.kafka: hosts: [192.168.140.130:9092] topic: nginxlog partition.round_robin: reachable_only: false required_acks: 1 compression: gzip max_message_size: 1048576 关闭 kibana setup.kibana: enabled: false4.4 Kafka 提前创建 topic重要在 Kafka 主机 192.168.140.130 执行# 创建 nginxlog 主题 kafka-topics.sh --bootstrap-server 192.168.140.130:9092 --create --topic nginxlog --partitions 3 --replication-factor 1 查看 topic 是否创建成功 kafka-topics.sh --bootstrap-server 192.168.140.130:9092 --list4.5 Filebeat 启动命令4.5.1 前台调试看报错cd /usr/local/filebeat ./filebeat -e -c filebeat.yml4.5.2 后台生产启动nohup ./filebeat -c filebeat.yml filebeat.log 21 4.5.3 查看进程ps -ef | grep filebeat4.5.4 停止 Filebeatpkill -f filebeat4.6 全套验证命令4.6.1 制造 Nginx 访问产生日志1. 使用 curl 命令访问 Nginx生成访问日志curl http://127.0.0.1/index.html curl http://127.0.0.1/404test2. Kafka 控制台消费者验证 Filebeat 生产者投递kafka-console-consumer.sh --bootstrap-server 192.168.140.130:9092 --topic nginxlog --from-beginning输出示例Filebeat 产出 JSON 格式{timestamp:2026-08-17T12:59:52.914Z,message:127.0.0.1 - - [17/Aug/2026:14:30:22 0800] \GET /index.html HTTP/1.1\ 200 1024 \-\ \Mozilla/5.0\ \-\}3. 查看 Nginx 原始日志tail -f /usr/local/nginx/logs/access.log4. 查看 Filebeat 运行日志tail -f /usr/local/filebeat/filebeat.log4.6.2 输出结果示例Kafka 控制台消费输出示例{ timestamp:2026-08-17T06:30:22.120Z, input:{type:filestream}, host:{name:kafka1}, message:127.0.0.1 - - [17/Aug/2026:14:30:22 0800] \GET /index.html HTTP/1.1\ 200 1024 \-\ \Mozilla/5.0\ \-\ }