Nginx + Lua + Kafka 实现数据日志收集

用户访问某些界面频率越高，那么此时可以推断当前商品是热门商品，所以我们可以收集商品详情页访问日志，收集方式如上图：

1:用户访问界面会经过Nginx
2:此时使用log_by_lua收集日志
3:log_by_lua将日志发送到Kafka

Nginx安装

Kafka安装

我们可以将我们需要查看的界面上传服务器，在nginx中配置好相关路径。配置在serve下。

		 #静态资源
        location ~ .*.(woff|ico|css|js|gif|jpg|jpeg|png)$ {
                root /usr/local/openrestyDir/pages;
        }

        #所有以msitems开始的请求都到路径下找详情页
        location /msitems/ {
                root /usr/local/openrestyDir/pages;
        }

Lua操作Kafka主要实现MQ消息发送和MQ消息消费，我们项目中主要实现发送消息，此时我们需要写一个Lua库或者借助其他Lua库，写一个Lua库成本比较大，可以借助github上开源的lua库https://github.com/doujiang24/lua-resty-kafka。

下载安装包：

将下载好的该库文件lua-resty-kafka-master上传至/usr/local/openresty，并解压unzip lua-resty-kafka-master.zip -d /usr/local/openresty。

nginx要想使用到该lua库，需要在nginx.conf中配置当前lua库文件路径，在http中配置如下：

lua_package_path "/usr/local/openresty/lua-resty-kafka-master/lib/?.lua;;";

用户访问界面，我们需要使用log_by_lua收集日志，我们先编写一个lua脚本实现日志收集，创建log.lua，脚本代码如下：

--引入json解析库
local cjson = require("cjson")
--kafka库
local producer = require "resty.kafka.producer"
--kafka的链接地址
local broker_list = {
      { host = "192.168.100.130", port = 9092 }
}
--生产者
local pro = producer:new(broker_list,{ producer_type="async"})

--用户IP
local headers=ngx.req.get_headers()
local ip=headers["X-REAL-IP"] or headers["X_FORWARDED_FOR"] or ngx.var.remote_addr or "0.0.0.0"

--消息内容
local logjson = {}
logjson["uri"]=ngx.var.uri
logjson["ip"]=ip
logjson["accesstime"]=os.date("%Y-%m-%d %H:%m:%S")

--发送消息
local offset, err = pro:send("mslogs", nil, cjson.encode(logjson))

if not ok then
	ngx.log(ngx.ERR, "kafka send err:", err)
	return
end

修改nginx.conf中界面访问的配置，使用log_by_lua_file向Kafka记录日志：

#所有以msitems开始的请求都到路径下找详情页
location /msitems/ {
root /usr/local/openrestyDir/pages;
# 日志收集
log_by_lua_file /usr/local/openresty/nginx/lua/logs.lua;
}

重启加载nginx配置(nginx -s reload)；

在Kafka界面打开消费者接受消息：

sh bin/kafka-console-consumer.sh --bootstrap-server 192.168.100.130:9092 --topic logs --from-beginning

访问http://192.168.100.130/msitems/1.html；

这个时候kafka会接收到相关消息，信息如下：

注意事项：

• 注意lua脚本中发送的消息是到那个topic中，Kafka中的消费者也需要监听该topic；

• 如果连接的是远程服务器的Kafka，需要确保Kafka是否支持远程连接。

  远程连接异常

接收到消息后，我们便可以将接收到的数据保存至数据库中。