【原创】运维基础之OpenResty(Nginx+Lua)+Kafka

lua_package_path "/usr/local/openresty/lualib/resty/kafka/?.lua;;";
lua_need_request_body on;
server {
listen ;
server_name testkafka;
location /test {
content_by_lua '
local testfile = "/tmp/test.log" local cjson = require "cjson"
local client = require "resty.kafka.client"
local producer = require "resty.kafka.producer" local broker_list = {
{ host = "127.0.0.1", port = }
} local topic = "test"
local key = "key"
local message = "halo world" -- usually we do not use this library directly
local cli = client:new(broker_list)
local brokers, partitions = cli:fetch_metadata(topic)
if not brokers then
ngx.say("fetch_metadata failed, err:", partitions)
end
--ngx.say("brokers: ", cjson.encode(brokers), "; partitions: ", cjson.encode(partitions)) -- sync producer_type
local p = producer:new(broker_list) local f = io.open(testfile, "a+")
f:write(topic .. ":" .. key .. ":" .. message .. "\\n")
f:close() local offset, err = p:send(topic, key, message)
if not offset then
ngx.say("send err:", err)
return
end
ngx.say("send success, offset: ", tonumber(offset)) -- this is async producer_type and bp will be reused in the whole nginx worker
local bp = producer:new(broker_list, { producer_type = "async" }) local ok, err = bp:send(topic, key, message)
if not ok then
ngx.say("send err:", err)
return
end ngx.say("host : ", ngx.var.host)
ngx.say("uri : ", ngx.var.uri)
ngx.say("args : ", ngx.var.args)
ngx.say("body : ", ngx.req.get_body_data())
ngx.say("client ip : ", ngx.var.remote_addr)
ngx.say("time : ", ngx.var.time_local)
ngx.say("send success, ok:", ok)
';
}
}