Nginx与Kafka整合方案

Nginx本身并不直接支持与Kafka的集成，但可以通过以下几种方式实现Nginx与Kafka的整合：

方案一：使用Nginx Lua模块 + Kafka客户端库

1. 安装OpenResty (包含Nginx和LuaJIT的增强版本)

   # Ubuntu/Debian
   sudo apt-get install -y openresty
   
   # CentOS/RHEL
   sudo yum install -y openresty
   

2. 安装Lua Kafka客户端库 (如lua-resty-kafka)

   luarocks install lua-resty-kafka
   

3. 配置Nginx (在nginx.conf或单独的配置文件中)

   http {
      lua_package_path "/path/to/lua-resty-kafka/lib/?.lua;;";
   
      server {
          location /log-to-kafka {
              content_by_lua_block {
                  local producer = require "resty.kafka.producer"
   
                  local broker_list = {
                      { host = "kafka1.example.com", port = 9092 },
                      { host = "kafka2.example.com", port = 9092 }
                  }
   
                  local p = producer:new(broker_list)
                  local ok, err = p:send("nginx-logs", nil, ngx.var.request_uri)
                  if not ok then
                      ngx.log(ngx.ERR, "kafka send err: ", err)
                      return ngx.exit(500)
                  end
                  ngx.say("message sent to kafka")
              }
          }
      }
   }
   

方案二：使用Nginx日志模块 + Kafka日志收集器

1. 配置Nginx访问日志

   http {
      log_format kafka_format '$remote_addr - $remote_user [$time_local] '
                              '"$request" $status $body_bytes_sent '
                              '"$http_referer" "$http_user_agent"';
   
      access_log /var/log/nginx/access.log kafka_format;
   }
   

2. 使用日志收集工具将日志发送到Kafka

   • Filebeat + Kafka: 配置Filebeat读取Nginx日志并发送到Kafka
   • Fluentd + Kafka: 使用Fluentd的kafka插件
   • Logstash + Kafka: 使用Logstash的kafka输出插件

   示例Filebeat配置(filebeat.yml):

   filebeat.inputs:
   - type: log
    paths:
      - /var/log/nginx/access.log
   
   output.kafka:
    hosts: ["kafka1:9092", "kafka2:9092"]
    topic: "nginx-logs"
    codec.format:
      string: '%{[message]}'
   

方案三：使用Nginx JavaScript模块 (njs)

1. 安装支持njs的Nginx版本

   # Ubuntu/Debian
   sudo apt-get install nginx-module-njs
   

2. 配置Nginx使用njs发送到Kafka

   load_module modules/ngx_http_js_module.so;
   
   http {
      js_import kafka from /etc/nginx/kafka.js;
   
      server {
          location /log {
              js_content kafka.logRequest;
          }
      }
   }
   

   kafka.js文件示例:

   function logRequest(r) {
      const kafka = require('kafka-node');
      const Producer = kafka.Producer;
      const client = new kafka.KafkaClient({kafkaHost: 'kafka:9092'});
      const producer = new Producer(client);
   
      producer.on('ready', function() {
          producer.send([{
              topic: 'nginx-logs',
              messages: JSON.stringify({
                  url: r.uri,
                  method: r.method,
                  headers: r.headersIn
              })
          }], function(err, data) {
              if (err) {
                  r.return(500, "Error sending to Kafka");
              } else {
                  r.return(200, "Logged to Kafka");
              }
          });
      });
   }
   
   export default {logRequest};
   

注意事项

1. 性能考虑: 直接在Nginx中集成Kafka客户端可能会影响性能，建议在高流量场景下使用日志收集器方案
2. 错误处理: 确保有适当的错误处理和重试机制
3. 安全性: 配置Kafka的认证和授权
4. 监控: 监控消息生产和消费情况

选择哪种方案取决于您的具体需求、性能要求和现有技术栈。