Nginx原生不支持直接推JSON日志至Kafka，主流生产方案有三：一是Nginx+rsyslog，轻量稳定、免日志轮转；二是Nginx+Logstash，字段丰富、支持Grok解析与条件过滤；三是OpenResty+lua-resty-kafka，低延迟、高吞吐、支持业务级控制。
直接在 Nginx 里把结构化 JSON 日志推到 Kafka，原生 Nginx 不支持——它没有内置 Kafka 生产者模块。但有三种主流、稳定、生产可用的方案，按适用场景和复杂度排序如下：
方案一：Nginx + rsyslog（轻量、稳定、免日志轮转）
适合对延迟不敏感、希望简化运维、彻底摆脱 logrotate 的场景。
- 确保 Nginx ≥ 1.10，启用 syslog 支持（默认已编译进多数发行版）
- 在
http块中定义 JSON 日志格式（注意引号和换行）：
log_format jsonlog escape=json '{'
'"time_iso8601":"$time_iso8601",'
'"remote_addr":"$remote_addr",'
'"host":"$host",'
'"request_uri":"$request_uri",'
'"status":$status,'
'"body_bytes_sent":$body_bytes_sent,'
'"request_time":$request_time,'
'"upstream_response_time":"$upstream_response_time",'
'"http_user_agent":"$http_user_agent",'
'"http_x_forwarded_for":"$http_x_forwarded_for"'
'}';
- 在
server或location中使用 syslog 输出（不写文件）：
access_log syslog:server=192.168.200.100:514,facility=local7,tag=nginx,jsonlog;
- 在 rsyslog 服务端配置
omkafka模块，将日志转发至 Kafka topic - 优势：零磁盘 IO、自动断连重试（rsyslog v8.20+）、天然支持日志分级与过滤
方案二：Nginx + Logstash（字段丰富、可过滤、兼容 ELK）
适合需保留原始 access.log、要 Grok 解析、做条件路由或对接 ES/Flink 的场景。
Wjs Uploading Video
下载
将一个或多个视频上传至YouTube。当用户想“上传到YouTube”“发YouTube”“批量上传”“upload to YouTube”“post videos to YouTube”或发布到YouTube时使用此技能。
- 同上配置 JSON 格式并输出到本地文件（如
/var/log/nginx/access.json） - Logstash input 使用
file插件，指定codec => "json"直接解析 - output 配置 Kafka 插件，支持批量、压缩、重试、分区键（如按 status 或 uri 分区）：
output {
kafka {
bootstrap_servers => ["192.168.200.139:9092", "192.168.200.140:9092"]
topic_id => "nginx-access"
codec => json { }
}
}
- 注意：Logstash 依赖
.sincedb记录读取位置，避免重复消费；建议关闭start_position => "beginning"（仅调试用）
方案三：OpenResty + lua-resty-kafka（低延迟、高吞吐、业务级控制）
适合对实时性要求高（毫秒级）、需在入口层做请求过滤/脱敏/打标、或已有 OpenResty 基础的团队。
- 安装 OpenResty（含 LuaJIT）及
lua-resty-kafka库 - 在
nginx.conf的http块中初始化 Kafka producer：
init_worker_by_lua_block {
local broker_list = {{"192.168.200.139", 9092}, {"192.168.200.140", 9092}}
local ok, err = kafka:new_producer(broker_list)
if not ok then
ngx.log(ngx.ERR, "failed to create kafka producer: ", err)
end
}
- 在
log_by_lua_block中构造 JSON 并异步发送（失败自动重试）：
log_by_lua_block {
local json = require "cjson"
local msg = json.encode({
ts = ngx.var.time_iso8601,
ip = ngx.var.remote_addr,
uri = ngx.var.request_uri,
status = tonumber(ngx.var.status),
rt = tonumber(ngx.var.request_time)
})
local ok, err = kafka:send("nginx-access", nil, msg)
if not ok then ngx.log(ngx.WARN, "kafka send failed: ", err) end
}
- 优势：无额外进程、支持自定义 key/partition、可结合 Redis 做限流或缓存判断
不推荐前端 AJAX 调用 /log 接口的方式——它本质是模拟日志，无法捕获 4xx/5xx 错误、不记录 upstream 时间、且增加客户端负担和 CORS 复杂度。
