OpenResty + Lua + Kafka 实现日志收集系统

一、Kafka 安装

官网下载 kafka_2.11-1.1.1.tgz，解压后，修改kafka的配置文件：config/server.properties

broker.id=0  # 在集群内必须惟一

advertised.host.name=192.168.10.100  # 配置对外IP地址，不然连接不上kafka

log.dirs=/data/kafka-logs  # 配置kafka的存储目录，包含kafka的日志和写入kafka的文件

zookeeper.connect=zk.test.com:2181  # 配置zookeeper的地址

二、启动kafka服务

nohup sh bin/kafka-server-start.sh config/server.properties > /data/kafka-logs/server.log 2>&1 &

启动kafka服务，并将服务端日志写入 server.log 文件

三、建立topic

bin/kafka-topics.sh --zookeeper zk.test.com:2181 --create --topic test1 --partitions 1 --replication-factor 1

建立名称为 test1 的topic

四、kafka测试

bin/kafka-console-producer.sh --broker-list zk.test.com:9092 --topic test1

建立一个消息生产者，输入消息则会发送到kafka

bin/kafka-console-consumer.sh --zookeeper zk.test.com:2181 --topic test1 --from-beginning

建立一个消费者，从topic头部开始消费信息

五、配置 lua-resty-kafka 

lua-resty-kafka 下载地址：https://github.com/doujiang24/lua-resty-kafka

因为需用到lua解析json，则还须要下载lua-cjson：https://github.com/openresty/lua-cjson/

将相应的配置导入到OpenResty的lua配置中

六、配置 OpenResty

server {
        listen    8088;
        location / {
            default_type text/html;
            content_by_lua '
                -- 引入lua全部api  
                local cjson = require "cjson"  
                local producer = require "resty.kafka.producer"  
                -- 定义kafka broker地址  
                local broker_list = {  
                    { host = "192.168.10.100", port = 9092 },  
                }  
                -- 定义json便于日志数据整理收集  
                local log_json = {}  
                log_json["uri"]=ngx.var.uri  
                log_json["args"]=ngx.var.args  
                log_json["host"]=ngx.var.host  
                log_json["request_body"]=ngx.var.request_body  
                log_json["remote_addr"] = ngx.var.remote_addr  
                log_json["remote_user"] = ngx.var.remote_user  
                log_json["time_local"] = ngx.var.time_local  
                log_json["status"] = ngx.var.status  
                log_json["body_bytes_sent"] = ngx.var.body_bytes_sent  
                log_json["http_referer"] = ngx.var.http_referer  
                log_json["http_user_agent"] = ngx.var.http_user_agent  
                log_json["http_x_forwarded_for"] = ngx.var.http_x_forwarded_for  
                log_json["upstream_response_time"] = ngx.var.upstream_response_time  
                log_json["request_time"] = ngx.var.request_time  
                -- 转换json为字符串  
                local message = cjson.encode(log_json);  
                -- 定义kafka异步生产者  
                local bp = producer:new(broker_list, { producer_type = "async" })  
                -- 发送日志消息,send第二个参数key,用于kafka路由控制:  
                -- key为nill(空)时，一段时间向同一partition写入数据  
                -- 指定key，按照key的hash写入到对应的partition  
                local ok, err = bp:send("test1", nil, message)  
   
                ngx.say("<br>", message)
                ngx.say("<br>kafka result:", ok)
                ngx.say("<br>kafka error:", err)
            ';
        }
    }

七、测试

访问 localhost:8088，则会将访问的日志写入kafka

八、kafka的一些命令

基础的命令都在kafka/bin 的目录下

kafka-server-start.sh 服务启动命令

kafka-topics.sh topic的命令，能够查看topic的状况，包括名称、分区等信息

kafka-console-producer.sh 生产者命令

kafka-console-consumer.sh 消费者名称