Kafka做日志收集

Posted 2022-12-02 Lossdate

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Kafka做日志收集相关的知识，希望对你有一定的参考价值。

一、简介

工作流程
1）用户的操作会由Web服务器进行响应
2）同时用户的操作也会使用ajax向nginx发送请求，Nginx用于收集用户的点击数据流
3）Nginx收集的日志数据使用ngx_kafka_module将数据发送到Kafka集群的主题中
4）只要数据保存到Kafka集群主题，后续就可以使用大数据组件进行实时计算或其他的处理了
架构
html + Nginx + ngx_kafka_module + Kafka
ngx_kafka_module下载：https://github.com/brg-liuwei/ngx_kafka_module
注：由于ngx_kafka_module只能接收POST请求，同时一般Web服务器不会和数据收集的Nginx在同一个域名，会涉及到跨域问题，可以在nginx中配置跨域来解决
环境
192.168.200.136 nodeA -> nginx & ngx_kafka_module
192.168.200.139 nodeB -> kafka
192.168.200.140 nodeC -> kafka
192.168.200.141 nodeD -> kafka

二、安装ngx_kafka_module

安装git
```
yum install -y git
```

安装librdkafka

mkdir /usr/local/kafka-module
cd /usr/local/kafka-module
git clone https://github.com/edenhill/librdkafka

cd /usr/local/kafka-module/librdkafka
./configure
make && make install

问题：c++ : 未找到命令
解决：

yum -y install gcc+ gcc-c++

下载ngx_kafka_module

cd /usr/local/kafka-module
git clone https://github.com/brg-liuwei/ngx_kafka_module

三、安装nginx (案例使用nginx-1.19.6.tar.gz)

解压

tar -zxvf nginx-1.19.6.tar.gz -C /usr/local

编译

cd /usr/local/nginx-1.19.6/
./configure --add-module=/usr/local/kafka-module/ngx_kafka_module

安装
```
make && make install
```
问题：make: *** 没有规则可以创建“default”需要的目标“build”。停止。
解决：
```
yum -y install gcc openssl openssl-devel pcre-devel zlib zlib-devel
```

配置环境变量

cd /etc/profile

#末尾追加
export NGINX_HOME=/usr/local/nginx
export PATH=$PATH:$NGINX_HOME/sbin

生效配置
```
. /etc/profile
```
启动
```
cd /usr/local/nginx/sbin 
./nginx
```
问题：./nginx: error while loading shared libraries: librdkafka.so.1: cannot open shared object file: No such file or directory
解决：
```
echo "/usr/local/lib" >> /etc/ld.so.conf
ldconfig
```
验证
```
浏览器访问对应的80口
```

四、kafka集群配置

安装和启动Kafka集群
基于JDK1.8的Kafka集群搭建

创建主题 tp_nginx_kafka

kafka-topics.sh --zookeeper nodeB:2181/myKafka --create --topic tp_nginx_kafka --partitions 3 --replication-factor 2

nodeA修改nginx配置文件

vim /usr/local/nginx/conf/nginx.conf

#在http下添加
#kafka配置
kafka;
#配置kafka broker,多个用逗号隔开
kafka_broker_list 192.168.200.139:9092,192.168.200.140:9092,192.168.200.141:9092;
server 
    #topic 消息转发到kafka
    location = /log.gif 
        #开启跨域
        add_header 'Access-Control-Allow-Origin' '*';
        add_header 'Access-Control-Allow-Credentials' 'true';
        kafka_topic tp_nginx_kafka;

重新加载nginx

cd /usr/local/nginx/sbin 
nginx -s reload

测试

#nodeB kafak启动消费
kafka-console-consumer.sh --bootstrap-server nodeB:9092 --topic tp_nginx_kafka

#nodeA 向kafka集群发送消息测试
curl nodeA/log.gif -d "this is a gift to kafka -> tp_nginx_kafka"

可以看到消息发送成功且集群收到消息

五、验证

页面 nginx-kafka.html

<!doctype html>
<html>
<head>
  <meta charSet="utf-8"/>
  <meta name="viewport" content="width=device-width, initial-scale=1.0, maximum-scale=5.0, minimum-scale=1.0"/>
  <title>nginx-kafka</title>
</head>
<body>
  <button onclick="sendkfk('signIn')">登录</button>
  <button onclick="sendkfk('signUp')">注册</button>
  <button onclick="sendkfk('click')">点击</button>
  <button onclick="sendkfk('like')">点赞</button>
  <button onclick="sendkfk('collect')">收藏</button>
  <button onclick="sendkfk('comment')">评论</button>
  <button onclick="sendkfk('send')">转发</button>
  
  <script type="text/javascript" src='https://cdn.bootcss.com/jquery/3.3.1/jquery.min.js'></script>
  <script>
    function sendkfk(currentAction)
        var info = ;
        info.userId = 123456;
		info.userName = 'Test';
        info.actTime = new Date();
        info.action = currentAction;
        info.code = "LEGEND-CEO";
        $.ajax(
            type: "post",
            url: "/log.gif",
            data: JSON.stringify(info),
            success: function(res)
            
        );
    
  </script>
</body>
</html>

上传到服务器

mkdir -p /usr/local/nginx/html/static/kafka-test
cd /usr/local/nginx/html/static/kafka-test/
rz

配置nginx.conf

vim /usr/local/nginx/conf/nginx.conf

server 
  listen   80;
  server_name  localhost;
  
  location / 
     root /usr/local/nginx/html/static/kafka-test/;
     index nginx-kafka.html;

重新加载nginx

cd /usr/local/nginx/sbin 
nginx -s reload

访问 http://192.168.200.136/static/kafka-test/nginx-kafka.html
点击功能按钮后在kafka上验证

以上是关于Kafka做日志收集的主要内容，如果未能解决你的问题，请参考以下文章

大数据篇：flume+kafka+spark stream+hbase做日志收集

mysql+canal+kafka做系统日志收集

elasticsearch＋kafka日志收集和分析以及分布式配置(附)

使用redis，kafka做为缓存来收集日志

kafka操作日志管理