亿级流量电商详情页系统实战-36.基于nginx+lua完成商品详情页访问流量实时上报kafka的开发

清疚 2022-10-30 07:27 157阅读 0赞

# 1.前言 #

在nginx这一层，接收到访问请求的时候，就把请求的流量上报发送给kafka。

这样的话，storm才能去消费kafka中的实时的访问日志，然后去进行缓存热数据的统计。

用得技术方案非常简单，从lua脚本直接创建一个kafka producer，发送数据到kafka。

# 2.配置 #

## 2.1下载ua-resty-kafka ##

![在这里插入图片描述][20210218162523403.png]

#cd /usr/servers/lualib/resty
    #wget https://github.com/doujiang24/lua-resty-kafka/archive/v0.07.zip
    #yum install -y unzip
    #unzip v0.07.zip
    #cp -rf lua-resty-kafka-0.07/lib/resty/kafka .

## 2.2 编写相关lua kafka producer ##

# vi /usr/lua/eshop/lua/product.lua
    
    local cjson = require("cjson")  
    local producer = require("resty.kafka.producer")  
    
    local broker_list = {   
        {  host = "192.168.135.135", port = 9092 },  
        {  host = "192.168.135.132", port = 9092 },  
        {  host = "192.168.135.136", port = 9092 }
    }
    
    local log_json = { }  
    log_json["request_module"] = "product_detail_info" 
    log_json["headers"] = ngx.req.get_headers()  
    log_json["uri_args"] = ngx.req.get_uri_args()  
    log_json["body"] = ngx.req.read_body()  
    log_json["http_version"] = ngx.req.http_version()  
    log_json["method"] =ngx.req.get_method() 
    log_json["raw_reader"] = ngx.req.raw_header()  
    log_json["body_data"] = ngx.req.get_body_data()  
    
    local message = cjson.encode(log_json);  
    
    local productId = ngx.req.get_uri_args()["productId"]
    
    local async_producer = producer:new(broker_list, {  producer_type = "async" })   
    local ok, err = async_producer:send("access-log", productId, message)  
    
    if not ok then  
        ngx.log(ngx.ERR, "kafka send err:", err)  
        return  
    end

## 2.3 使nginx配置生效 ##

--增加 resolver 8.8.8.8
    # vi /usr/servers/nginx/conf/nginx.conf 
    http { 
        resolver 8.8.8.8;
        include       mime.types;
        default_type  application/octet-stream;
    
    #/usr/servers/nginx/sbin/nginx -s reload

## 2.4 配置kafka ##

--设置advertised.host.name
    # vi /usr/local/kafka/config/server.properties
    
    --重启kafka进程
    # jps
    1215 Kafka
    #kill 1215
    #nohup /usr/local/kafka/bin/kafka-server-start.sh /usr/local/kafka/config/server.properties &
    
    --增加access-log topic
    # /usr/local/kafka/bin/kafka-topics.sh --zookeeper 192.168.135.135:2181,192.168.135.132:2181,192.168.135.136:2181 --topic access-log --replication-factor 1 --partitions 1 --create
    
    # /usr/local/kafka/bin/kafka-console-consumer.sh --zookeeper 192.168.135.135:2181,192.168.135.132:2181,192.168.135.136:2181 --topic access-log --from-beginning

# 3.测试 #

1.  启动eshop-cache缓存服务
2.  http://192.168.135.135/eshop?method=product&productId=1&shopId=1

[20210218162523403.png]: /images/20221024/7c55e04160b840fc9452903278cc556f.png