当前位置：首页 > news >正文

基于kafka的分布式日志收集平台项目（第三部分）

news 2025/7/10 8:16:39

续第一部分最后面没精讲部分和后面剩余部分

nginx反向代理集群搭建

安装nginx

输入yum install nginx -y

如果报错则输入以下两串命令

yum install epel-release -y
yum install nginx -y

##以下和以上这些操作都在kafka1里面进行，至于为什么要进行下面这些操作，是因为咱只用三台机器，没有那么多机器可用，所以我们在kafka2和kafka3上面又部署了两个python，让kafka1里面的nginx代理到下面两个python里面

编辑配置文件 /etc/nginx/conf.d/sc.conf（进入目录 /etc/nginx/conf.d/，有sc.conf文件就直接编辑，没有就新建）

upstream flask {
server kafka2;
server kafka3;

}

server {
server_name www.sc.com;
location / {
proxy_pass http://flask;
}

}

启动nginx：

systemctl start nginx

后端flask程序（这一部分要在kafka2和kafka3两个里面都执行，kafka1不执行）

安装flask环境：

yum install python3 -y
pip3 install flask -i https://pypi.tuna.tsinghua.edu.cn/simple

编辑/opt/python-flask/app.py文件（新建文件夹python-flask再新建app.py文件并编辑，直接把下面的文本复制进去）：

from flask import Flask

app = Flask(__name__)

@app.route("/")
def index():
return "this is flask web kafka2" #在kafka3节点里这个里面的kafka2就改为kafka3

app.run(host = "0.0.0.0")

启动flask：

python3 app.py

出现以下结果就算成功，并且不能中断

然后可以在网页输入IP地址：5000

例（以我自己为例）：

#kafka2

输入 192.168.126.175：5000

然后输出结果

this is flask web kafka2

#kafka3

输入 192.168.126.176：5000

然后输出结果

this is flask web kafka3

#kafka1（可以一直按回车键，里面内容的kafka2和kafka3会一直变换）

输入 192.168.126.173：5000

然后输出结果

this is flask web kafka2

部署filebeat

可以先了解一下何为filebeat

一篇文章搞懂filebeat（ELK） - 一寸HUI - 博客园https://www.cnblogs.com/zsql/p/13137833.html

安装：

1、rpm --import https://packages.elastic.co/GPG-KEY-elasticsearch
2、编辑 vim /etc/yum.repos.d/fb.repo
[elastic-7.x]
name=Elastic repository for 7.x packages
baseurl=https://artifacts.elastic.co/packages/7.x/yum
gpgcheck=1
gpgkey=https://artifacts.elastic.co/GPG-KEY-elasticsearch
enabled=1
autorefresh=1
type=rpm-md
3、yum安装
yum install filebeat -y

rpm -qa |grep filebeat #可以查看filebeat有没有安装 rpm -qa 是查看机器上安装的所有软件包
rpm -ql filebeat 查看filebeat安装到哪里去了，牵扯的文件有哪些

配置，修改配置文件/etc/filebeat/filebeat.yml：

filebeat.inputs:
- type: log
# Change to true to enable this input configuration.
enabled: true
# Paths that should be crawled and fetched. Glob based paths.
paths:
- /var/log/nginx/access.log
- /var/log/nginx/error.log
#==========------------------------------kafka-----------------------------------
output.kafka:
hosts: ["192.168.20.161:9092","192.168.20.162:9092","192.168.20.163:9092"]
topic: nginxlog
keep_alive: 10s

不对的就修改，没有的就增加

ip换上自己的ip

创建主题：

cd /opt/kafka_2.13-3.6.1
bin/kafka-topics.sh --create --bootstrap-server kafka3:9092 --replication-factor 3 --partitions 3 --topic nginxlog

启动服务：

systemctl start filebeat
systemctl enable filebeat #设置开机自启

可以输入命令ps aux|grep filebeat 检查一下filebeat服务是否启动

#创建生产者，发送消息，测试用
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

用上述两条命令再创建生产者nginxlog 再创建消费者来消费（图有点不美观，随便输入了点啥，可以看到生产者输入的内容可以由消费者输出）