nginx+lua 实现请求流量上报kafka

文章目录

[隐藏]

• 环境依赖
• 代码实现

环境依赖

前面26、27、28讲到的博文环境即可，上报kafka ，只需在应用层nginx上操作（192.168.0.16,192.168.0.17）

请求上报kafka 其实很简单，大致思路是：

• 下载lua-resty-kafka，提供lua 操作kafka的方法类库
• lua 获取nginx 请求参数，组装上报对象
• 上报对象 encode cjson 编码
• lua kakfa 上报即可

代码实现

• 引入 lua-resty-kafka 类库

yum install -y unzip  cd /usr/local/servers/ && wget https://github.com/doujiang24/lua-resty-kafka/archive/master.zip  unzip master.zip  cp -rf lua-resty-kafka-master/lib/resty/kafka /usr/local/test/lualib/resty/  /usr/local/servers/nginx/sbin/nginx -s reload  

• lua 获取请求，组装上报对象，encode对象并上报（注意：以下代码只对流量上报代码进行注释，其他代码参考 前面 28 “分发层 + 应用层” 双层nginx 架构 之 应用层实现）

vim /usr/local/test/lua/test.lua  

代码如下：

// 引入 kafka 生产者 类库  local producer = require("resty.kafka.producer")  // 引入json 解析类库  local cjson = require("cjson")  // 构造kafka 集群节点 broker  local broker_list = {      { host = "192.168.0.16", port = 9092},      { host = "192.168.0.17", port = 9092},      { host = "192.168.0.18", port = 9092}  }  // 定义上报对象  local log_obj = {}  // 自定义模块名称  log_obj["request_module"] = "product_detail_info"  // 获取请求头信息  log_obj["headers"] = ngx.req.get_headers()  // 获取请求uri 参数  log_obj["uri_args"] = ngx.req.get_uri_args()  // 获取请求body  log_obj["body"] = ngx.req.read_body()  // 获取请求的http协议版本  log_obj["http_version"] = ngx.req.http_version()  // 获取请求方法  log_obj["method"] = ngx.req.get_method()  // 获取未解析的请求头字符串  log_obj["raw_reader"] = ngx.req.raw_header()  // 获取解析的请求body体内容字符串  log_obj["body_data"] = ngx.req.get_body_data()  // 上报对象json 字符串编码  local message = cjson.encode(log_obj)    local uri_args = ngx.req.get_uri_args()  local product_id = uri_args["productId"]  local shop_id = uri_args["shopId"]  // 创建kafka producer 连接对象，producer_type = "async" 异步  local async_producer = producer:new(broker_list, {producer_type = "async"})  // 请求上报kafka，kafka 生产者发送数据，async_prodecer:send（a，b，c），a : 主题名称，b：分区（保证相同id，全部到相同的kafka node 去，并且顺序一致），c：消息（上报数据）  local ok, err = async_producer:send("access-log", product_id, message)  // 上报异常处理  if not ok then     ngx.log(ngx.ERR, "kafka send err:", err)     return  end  local cache_ngx = ngx.shared.test_cache  local product_cache_key = "product_info_"..product_id  local shop_cache_key = "shop_info_"..shop_id  local product_cache = cache_ngx:get(product_cache_key)  local shop_cache = cache_ngx:get(shop_cache_key)  if product_cache == "" or product_cache == nil then        local http = require("resty.http")        local httpc = http.new()          local resp, err = httpc:request_uri("http://192.168.0.3:81",{          method = "GET",              path = "/getProductInfo?product" or shop_cache == nil then        local http = require("resty.http")        local httpc = http.new()        local resp, err = httpc:request_uri("http://192.168.0.3:81",{          method = "GET",              path = "/getShopInfo?shopresty.template")  template.render("product.html", context)  

• 配置nginx DNS resolver实例，避免 DNS 解析失败

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

在 http 部分添加以下内容，如下图：

resolver 8.8.8.8  

配置nginx dns resolver
（注：以上操作 nginx 应用服务器（192.168.0.16,192.168.0.17）都需要进行）

• 配置 kafka advertised.host.name 参数（避免通过机器名无法找到对应的机器）（所有kafka 节点都配置）

advertised.host.name = 本机ip

vim /usr/local/kafka/config/server.properties  

配置advertised.host.name

• nginx 校验 及 重载

/usr/local/servers/nginx/sbin/nginx -t && /usr/local/servers/nginx/sbin/nginx -s reload  

• 启动kafka(确保 zookeeper 已启动)

cd /usr/local/kafka && nohup bin/kafka-server-start.sh config/server.properties &  

• kafka 中创建 access-log 主题

cd cd /usr/local/kafka && bin/kafka-topics.sh –zookeeper my-cache1:2181,my-cache2:2181,my-cache3:2181 –topic access-log –replication-factor 1 –partitions 1 –create  

• 打开kafka consumer 查看数据

bin/kafka-console-consumer.sh –zookeeper my-cache1:2181,my-cache2:2181,my-cache3:2181 –topic access-log –from-beginning  

• 浏览器请求nginx

nginx请求

shell 打开kafka 消费端查看请求上报kafka 数据

完毕，利用nginx + lua 实现请求流量上报kafka就到这里了。

以上就是本章内容，如有不对的地方，请多多指教，谢谢！

原文出处：96nian -> http://www.96nian.com/4026.html