lua-resty-kafka配置文档
参考网址：https://github.com/doujiang24/lua-resty-kafka
一、例子
content_by_lua '
-- 引入lua所有api
local cjson = require "cjson"
local producer = require "resty.kafka.producer"
-- 定义kafka broker地址，ip需要和kafka的host.name配置一致
local broker_list = {
{ host = "192.168.101.223", port = 9092 },
{ host = "192.168.101.224", port = 9092 }
}
local key = "key"
local message = "halo world"
local error_handle = function (topic, partition_id, queue, index, err, retryable)
ngx.log(ngx.ERR, "failed to send to kafka, topic: ", topic, "; partition_id: ", partition_id, "; retryable: ", retryable)
end
local p = producer:new(broker_list, { producer_type = "async", max_retry = 1, batch_num = 1, error_handle = error_handle })
local ok, err = p:send("test", key, message)
if not ok then
ngx.say("send err:", err)
return
end
ngx.say("send ok:", ok)
p:flush()
';
二、语法
New
p = producer:new(broker_list, producer_config?, cluster_name?)
broker_list：客户端列表
local broker_list = {
{ host = "192.168.101.223", port = 9092 },
{ host = "192.168.101.224", port = 9092 }
}
producer_config?：可选参数
producer_type：生产者类型，同步或者异步，"async" or "sync"
request_timeout：请求超时，默认是 2000 ms
required_acks：请求应答，不能为0，默认是1
max_retry：信息发送重试最大次数，默认是3
retry_backoff：信息发送重试补偿，默认100ms
partitioner：选择分区从键和分区num的分区。
local partitioner = function (key, partition_num, correlation_id) end
默认
local function default_partitioner(key, num, correlation_id)local id = key and crc32(key) or correlation_id
-- partition_id is continuous and start from 0
return id % num
end
！！！以下为缓冲区参数，只有producer_type= "async"有效
flush_time：队列最大缓存时间，默认1000ms
batch_num：队列最大批次数量，默认200
batch_size：保存缓存大小，默认是1M（最大可以为2M），需要小心，这个跟kafka配置有关，socket.request.max.byts为2-10k
max_buffering：队列最大缓存大小，默认50000
error_handle：错误处理，当缓冲区发送到kafka错误时处理数据。
error_handle = function (topic, partition_id, message_queue, index, err, retryable) end
失败的消息队列如{ key1, msg1, key2, msg2 },键值key在消息队列中是为””相当与orign中的nil，
Index为消息队列的长度
Retryable为true时，这意味着kafka服务器肯定没有提交这些消息。可以尝试重试发送消息，
暂时不支持压缩
cluster_name?：可选参数
指定集群的名称，默认是1（该参数是一个数值），当您有两个或多个kafka集群时，您可以指定不同的名称，只有producer_type = "async"有效
Send
ok, err = p:send(topic, key, message)
同步模式时 ：
如果成功，返回当前代理和分区的偏移量(** cdata: LL **)。如果出现错误，返回nil，用字符串描述错误
异步模式时 ：
消息将首先写入缓冲区。当缓冲区超过batch_num时，它将发送到kafka服务器，或者每个flush_time刷新缓冲区。