Nginx 怎么在 Nginx 中配置将结构化 JSON 日志推送到 Kafka 实时流处理系统-Nginx-PHP中文网

nginx原生不支持直接推json日志至kafka，主流生产方案有三：一是nginx+rsyslog，轻量稳定、免日志轮转；二是nginx+logstash，字段丰富、支持grok解析与条件过滤；三是openresty+lua-resty-kafka，低延迟、高吞吐、支持业务级控制。

nginx 怎么在 nginx 中配置将结构化 json 日志推送到 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
}