Nginx Stream 模块开发：从入门到实战指南_Linux

1. 引言

nginx 的 stream 模块（ngx_stream_module）是 nginx 1.9.0 引入的 tcp/udp 代理核心框架，它使得 nginx 不仅能处理 http 流量，还能代理 mysql、redis、mqtt 等四层协议。与 http 模块不同，stream 模块工作在传输层，没有请求/响应概念，而是基于连接和会话。本文将从源码层面讲解 stream 模块的架构与开发方法，并通过一个完整的自定义模块示例，带你掌握 stream 模块的开发流程。

2. stream 模块架构概览

2.1 核心数据结构

stream 模块的核心结构体定义在 src/stream/ngx_stream.h 中：

// 核心上下文结构
typedef struct {
    void                        **main_conf;      // 主配置
    void                        **srv_conf;       // server 级别配置
    ngx_stream_session_t         *session;        // 当前会话
} ngx_stream_conf_ctx_t;
// 会话结构
struct ngx_stream_session_s {
    void                        **ctx;            // 模块上下文
    ngx_connection_t             *connection;     // 客户端连接
    ngx_connection_t             *upstream;       // 上游连接
    ngx_stream_variable_value_t  *variables;      // 变量
    ngx_stream_headers_in_t      *headers_in;     // 输入头部
    void                        **main_conf;
    void                        **srv_conf;
    ngx_log_t                    *log;
    unsigned                      ssl:1;
    unsigned                      proxy_protocol:1;
    // ...
};

2.2 模块生命周期

stream 模块的生命周期与 http 模块类似，但更简洁：

配置解析阶段：读取 stream {} 块中的指令，调用模块的 create_main_conf、create_srv_conf 和 preconfiguration 回调。
初始化阶段：init_main_conf 和 init_worker 回调。
会话处理阶段：当新连接到达时，调用模块的 handler 回调处理会话。
销毁阶段：exit_master 和 exit_worker 回调。

3. 模块开发核心接口

3.1 模块结构体定义

每个 stream 模块必须定义一个 ngx_stream_module_t 结构体：

typedef struct {
    ngx_int_t   (*preconfiguration)(ngx_conf_t *cf);
    ngx_int_t   (*postconfiguration)(ngx_conf_t *cf);
    void       *(*create_main_conf)(ngx_conf_t *cf);
    char       *(*init_main_conf)(ngx_conf_t *cf, void *conf);
    void       *(*create_srv_conf)(ngx_conf_t *cf);
    char       *(*merge_srv_conf)(ngx_conf_t *cf, void *parent, void *child);
} ngx_stream_module_t;

3.2 会话处理函数

stream 模块的核心处理函数签名如下：

typedef void (*ngx_stream_handler_pt)(ngx_stream_session_t *s);

与 http 模块返回状态码不同，stream 模块的处理函数是 void 类型，通过设置 s->connection->write->handler 或直接调用上游连接函数来驱动后续处理。

3.3 配置指令注册

使用 ngx_stream_command_t 数组注册配置指令：

typedef struct {
    ngx_str_t                    name;       // 指令名称
    ngx_uint_t                   type;       // 指令类型
    char               *(*set)(ngx_conf_t *cf, ngx_command_t *cmd, void *conf);
    ngx_uint_t                   conf;       // 配置级别
    ngx_uint_t                   offset;     // 偏移量
    void                        *post;       // 后处理函数
} ngx_stream_command_t;

4. 实战：开发一个 stream 日志模块

下面我们开发一个名为 ngx_stream_log_module 的 stream 模块，它能在每个 tcp 连接关闭时记录连接信息（客户端 ip、端口、连接时长、传输字节数等）到自定义日志文件。

4.1 模块头文件与结构体定义

// ngx_stream_log_module.h
#ifndef _ngx_stream_log_module_h_included_
#define _ngx_stream_log_module_h_included_
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_stream.h>
typedef struct {
    ngx_str_t    log_file;       // 日志文件路径
    ngx_flag_t   log_connect;    // 是否记录连接事件
    ngx_flag_t   log_disconnect; // 是否记录断开事件
} ngx_stream_log_srv_conf_t;
typedef struct {
    ngx_file_t   file;           // 日志文件句柄
    ngx_log_t    log;            // nginx 日志对象
} ngx_stream_log_ctx_t;
#endif

4.2 模块主文件

// ngx_stream_log_module.c
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_stream.h>
#include "ngx_stream_log_module.h"
// 模块上下文索引
static ngx_int_t ngx_stream_log_module_index;
// 前向声明
static void *ngx_stream_log_create_srv_conf(ngx_conf_t *cf);
static char *ngx_stream_log_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child);
static ngx_int_t ngx_stream_log_init(ngx_conf_t *cf);
static void ngx_stream_log_session_handler(ngx_stream_session_t *s);
static void ngx_stream_log_write_log(ngx_stream_session_t *s, ngx_int_t event_type);
// 配置指令
static ngx_command_t ngx_stream_log_commands[] = {
    { ngx_string("stream_log_file"),
      ngx_stream_srv_conf | ngx_conf_take1,
      ngx_conf_set_str_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_file),
      null },
    { ngx_string("stream_log_connect"),
      ngx_stream_srv_conf | ngx_conf_flag,
      ngx_conf_set_flag_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_connect),
      null },
    { ngx_string("stream_log_disconnect"),
      ngx_stream_srv_conf | ngx_conf_flag,
      ngx_conf_set_flag_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_disconnect),
      null },
    ngx_null_command
};
// 模块上下文
static ngx_stream_module_t ngx_stream_log_module_ctx = {
    null,                                  // preconfiguration
    ngx_stream_log_init,                   // postconfiguration
    null,                                  // create_main_conf
    null,                                  // init_main_conf
    ngx_stream_log_create_srv_conf,        // create_srv_conf
    ngx_stream_log_merge_srv_conf          // merge_srv_conf
};
// 模块定义
ngx_module_t ngx_stream_log_module = {
    ngx_module_v1,
    &ngx_stream_log_module_ctx,            // module context
    ngx_stream_log_commands,               // module directives
    ngx_stream_module,                     // module type
    null,                                  // init master
    null,                                  // init module
    null,                                  // init process
    null,                                  // init thread
    null,                                  // exit thread
    null,                                  // exit process
    null,                                  // exit master
    ngx_module_v1_padding
};

4.3 配置创建与合并

static void *
ngx_stream_log_create_srv_conf(ngx_conf_t *cf)
{
    ngx_stream_log_srv_conf_t *conf;
    conf = ngx_pcalloc(cf->pool, sizeof(ngx_stream_log_srv_conf_t));
    if (conf == null) {
        return null;
    }
    conf->log_file = ngx_null_string;
    conf->log_connect = ngx_conf_unset;
    conf->log_disconnect = ngx_conf_unset;
    return conf;
}
static char *
ngx_stream_log_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child)
{
    ngx_stream_log_srv_conf_t *prev = parent;
    ngx_stream_log_srv_conf_t *conf = child;
    ngx_conf_merge_str_value(conf->log_file, prev->log_file, "");
    ngx_conf_merge_value(conf->log_connect, prev->log_connect, 1);
    ngx_conf_merge_value(conf->log_disconnect, prev->log_disconnect, 1);
    return ngx_conf_ok;
}

4.4 初始化与会话处理

static ngx_int_t
ngx_stream_log_init(ngx_conf_t *cf)
{
    ngx_stream_core_main_conf_t *cmcf;
    ngx_stream_handler_pt       *h;
    // 获取核心模块的主配置
    cmcf = ngx_stream_conf_get_module_main_conf(cf, ngx_stream_core_module);
    // 注册会话处理函数到处理链
    h = ngx_array_push(&cmcf->phases[ngx_stream_content_phase].handlers);
    if (h == null) {
        return ngx_error;
    }
    *h = ngx_stream_log_session_handler;
    // 获取模块索引
    ngx_stream_log_module_index = ngx_stream_get_module_index(cf, &ngx_stream_log_module);
    return ngx_ok;
}
static void
ngx_stream_log_session_handler(ngx_stream_session_t *s)
{
    ngx_stream_log_srv_conf_t *lscf;
    ngx_stream_log_ctx_t      *ctx;
    // 获取 server 级别配置
    lscf = ngx_stream_get_module_srv_conf(s, ngx_stream_log_module);
    if (lscf->log_file.len == 0) {
        // 未配置日志文件，跳过
        goto next;
    }
    // 创建模块上下文
    ctx = ngx_pcalloc(s->connection->pool, sizeof(ngx_stream_log_ctx_t));
    if (ctx == null) {
        goto next;
    }
    ngx_stream_set_module_ctx(s, ngx_stream_log_module, ctx);
    // 打开日志文件（首次连接时）
    if (!ctx->file.fd) {
        ctx->file.name = lscf->log_file;
        ctx->file.log = s->connection->log;
        ctx->file.fd = ngx_open_file(ctx->file.name.data,
                                     ngx_file_append,
                                     ngx_file_create_or_open,
                                     ngx_file_default_access);
        if (ctx->file.fd == ngx_invalid_file) {
            ngx_log_error(ngx_log_err, s->connection->log, ngx_errno,
                          "stream_log: failed to open log file \"%v\"",
                          &lscf->log_file);
            goto next;
        }
    }
    // 记录连接事件
    if (lscf->log_connect) {
        ngx_stream_log_write_log(s, 0);  // 0 = connect
    }
    // 注册连接关闭回调
    s->connection->log->handler = ngx_stream_log_write_log;
    s->connection->log->data = s;
next:
    // 调用下一个处理函数
    ngx_stream_core_run_phases(s);
}

4.5 日志写入函数

static void
ngx_stream_log_write_log(ngx_stream_session_t *s, ngx_int_t event_type)
{
    u_char                      *p, *last;
    ngx_stream_log_ctx_t        *ctx;
    ngx_stream_log_srv_conf_t   *lscf;
    struct timeval               tv;
    ngx_time_t                  *tp;
    ngx_connection_t            *c;
    c = s->connection;
    lscf = ngx_stream_get_module_srv_conf(s, ngx_stream_log_module);
    ctx = ngx_stream_get_module_ctx(s, ngx_stream_log_module);
    if (ctx == null || ctx->file.fd == ngx_invalid_file) {
        return;
    }
    // 分配缓冲区
    last = ngx_pnalloc(c->pool, ngx_max_error_str);
    if (last == null) {
        return;
    }
    p = last;
    // 获取当前时间
    tp = ngx_timeofday();
    ngx_gmtime(tp->sec, &tp->gmtime);
    // 格式化日志行
    p = ngx_sprintf(p, "%4d-%02d-%02d %02d:%02d:%02d ",
                    tp->gmtime.ngx_tm_year, tp->gmtime.ngx_tm_mon,
                    tp->gmtime.ngx_tm_mday, tp->gmtime.ngx_tm_hour,
                    tp->gmtime.ngx_tm_min, tp->gmtime.ngx_tm_sec);
    p = ngx_sprintf(p, "client=%v:%d ", &c->addr_text, c->local_sockaddr ? ntohs(((struct sockaddr_in *)c->local_sockaddr)->sin_port) : 0);
    if (event_type == 0) {
        p = ngx_sprintf(p, "event=connect");
    } else {
        p = ngx_sprintf(p, "event=disconnect bytes_sent=%o bytes_received=%o duration=%dms",
                        c->sent, c->received,
                        (ngx_int_t)((ngx_time() - s->start_sec) * 1000));
    }
    *p++ = '\n';
    // 写入文件
    ngx_write_fd(ctx->file.fd, last, p - last);
}

5. 编译与配置

5.1 编译模块

将上述源码放在 nginx-1.xx.x/src/stream/ngx_stream_log_module/ 目录下，然后通过动态模块方式编译：

# 配置动态模块
./configure --add-dynamic-module=src/stream/ngx_stream_log_module
# 编译
make modules
# 编译后得到 objs/ngx_stream_log_module.so

5.2 nginx 配置示例

# 加载动态模块
load_module modules/ngx_stream_log_module.so;
stream {
    server {
        listen 3306;  # mysql 代理
        proxy_pass db_backend:3306;
        # 自定义日志配置
        stream_log_file /var/log/nginx/stream_access.log;
        stream_log_connect on;
        stream_log_disconnect on;
    }
    server {
        listen 6379;  # redis 代理
        proxy_pass redis_backend:6379;
        stream_log_file /var/log/nginx/stream_redis.log;
        stream_log_connect on;
        stream_log_disconnect on;
    }
}

5.3 日志输出示例

2026-07-17 10:23:45 client=192.168.1.100:54321 event=connect
2026-07-17 10:24:12 client=192.168.1.100:54321 event=disconnect bytes_sent=1024 bytes_received=2048 duration=27000ms

6. 完整实例代码

以下是完整的模块源码，可直接编译使用：

// ============================================================
// ngx_stream_log_module.h
// ============================================================
#ifndef _ngx_stream_log_module_h_included_
#define _ngx_stream_log_module_h_included_
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_stream.h>
typedef struct {
    ngx_str_t    log_file;
    ngx_flag_t   log_connect;
    ngx_flag_t   log_disconnect;
} ngx_stream_log_srv_conf_t;
typedef struct {
    ngx_file_t   file;
} ngx_stream_log_ctx_t;
#endif
// ============================================================
// ngx_stream_log_module.c
// ============================================================
#include <ngx_config.h>
#include <ngx_core.h>
#include <ngx_stream.h>
#include "ngx_stream_log_module.h"
static ngx_int_t ngx_stream_log_module_index;
static void *ngx_stream_log_create_srv_conf(ngx_conf_t *cf);
static char *ngx_stream_log_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child);
static ngx_int_t ngx_stream_log_init(ngx_conf_t *cf);
static void ngx_stream_log_session_handler(ngx_stream_session_t *s);
static void ngx_stream_log_write_log(ngx_stream_session_t *s, ngx_int_t event_type);
static ngx_command_t ngx_stream_log_commands[] = {
    { ngx_string("stream_log_file"),
      ngx_stream_srv_conf | ngx_conf_take1,
      ngx_conf_set_str_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_file),
      null },
    { ngx_string("stream_log_connect"),
      ngx_stream_srv_conf | ngx_conf_flag,
      ngx_conf_set_flag_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_connect),
      null },
    { ngx_string("stream_log_disconnect"),
      ngx_stream_srv_conf | ngx_conf_flag,
      ngx_conf_set_flag_slot,
      ngx_stream_srv_conf_offset,
      offsetof(ngx_stream_log_srv_conf_t, log_disconnect),
      null },
    ngx_null_command
};
static ngx_stream_module_t ngx_stream_log_module_ctx = {
    null,
    ngx_stream_log_init,
    null,
    null,
    ngx_stream_log_create_srv_conf,
    ngx_stream_log_merge_srv_conf
};
ngx_module_t ngx_stream_log_module = {
    ngx_module_v1,
    &ngx_stream_log_module_ctx,
    ngx_stream_log_commands,
    ngx_stream_module,
    null, null, null, null, null, null, null,
    ngx_module_v1_padding
};
static void *
ngx_stream_log_create_srv_conf(ngx_conf_t *cf)
{
    ngx_stream_log_srv_conf_t *conf;
    conf = ngx_pcalloc(cf->pool, sizeof(ngx_stream_log_srv_conf_t));
    if (conf == null) return null;
    conf->log_file = ngx_null_string;
    conf->log_connect = ngx_conf_unset;
    conf->log_disconnect = ngx_conf_unset;
    return conf;
}
static char *
ngx_stream_log_merge_srv_conf(ngx_conf_t *cf, void *parent, void *child)
{
    ngx_stream_log_srv_conf_t *prev = parent;
    ngx_stream_log_srv_conf_t *conf = child;
    ngx_conf_merge_str_value(conf->log_file, prev->log_file, "");
    ngx_conf_merge_value(conf->log_connect, prev->log_connect, 1);
    ngx_conf_merge_value(conf->log_disconnect, prev->log_disconnect, 1);
    return ngx_conf_ok;
}
static ngx_int_t
ngx_stream_log_init(ngx_conf_t *cf)
{
    ngx_stream_core_main_conf_t *cmcf;
    ngx_stream_handler_pt       *h;
    cmcf = ngx_stream_conf_get_module_main_conf(cf, ngx_stream_core_module);
    h = ngx_array_push(&cmcf->phases[ngx_stream_content_phase].handlers);
    if (h == null) return ngx_error;
    *h = ngx_stream_log_session_handler;
    ngx_stream_log_module_index = ngx_stream_get_module_index(cf, &ngx_stream_log_module);
    return ngx_ok;
}
static void
ngx_stream_log_session_handler(ngx_stream_session_t *s)
{
    ngx_stream_log_srv_conf_t *lscf;
    ngx_stream_log_ctx_t      *ctx;
    lscf = ngx_stream_get_module_srv_conf(s, ngx_stream_log_module);
    if (lscf->log_file.len == 0) goto next;
    ctx = ngx_pcalloc(s->connection->pool, sizeof(ngx_stream_log_ctx_t));
    if (ctx == null) goto next;
    ngx_stream_set_module_ctx(s, ngx_stream_log_module, ctx);
    if (!ctx->file.fd) {
        ctx->file.name = lscf->log_file;
        ctx->file.log = s->connection->log;
        ctx->file.fd = ngx_open_file(ctx->file.name.data,
                                     ngx_file_append,
                                     ngx_file_create_or_open,
                                     ngx_file_default_access);
        if (ctx->file.fd == ngx_invalid_file) {
            ngx_log_error(ngx_log_err, s->connection->log, ngx_errno,
                          "stream_log: failed to open log file \"%v\"",
                          &lscf->log_file);
            goto next;
        }
    }
    if (lscf->log_connect) {
        ngx_stream_log_write_log(s, 0);
    }
    s->connection->log->handler = ngx_stream_log_write_log;
    s->connection->log->data = s;
next:
    ngx_stream_core_run_phases(s);
}
static void
ngx_stream_log_write_log(ngx_stream_session_t *s, ngx_int_t event_type)
{
    u_char                     *p, *last;
    ngx_stream_log_ctx_t       *ctx;
    ngx_stream_log_srv_conf_t  *lscf;
    ngx_connection_t           *c;
    ngx_time_t                 *tp;
    c = s->connection;
    lscf = ngx_stream_get_module_srv_conf(s, ngx_stream_log_module);
    ctx = ngx_stream_get_module_ctx(s, ngx_stream_log_module);
    if (ctx == null || ctx->file.fd == ngx_invalid_file) return;
    last = ngx_pnalloc(c->pool, ngx_max_error_str);
    if (last == null) return;
    p = last;
    tp = ngx_timeofday();
    ngx_gmtime(tp->sec, &tp->gmtime);
    p = ngx_sprintf(p, "%4d-%02d-%02d %02d:%02d:%02d ",
                    tp->gmtime.ngx_tm_year, tp->gmtime.ngx_tm_mon,
                    tp->gmtime.ngx_tm_mday, tp->gmtime.ngx_tm_hour,
                    tp->gmtime.ngx_tm_min, tp->gmtime.ngx_tm_sec);
    p = ngx_sprintf(p, "client=%v ", &c->addr_text);
    if (event_type == 0) {
        p = ngx_sprintf(p, "event=connect");
    } else {
        p = ngx_sprintf(p, "event=disconnect sent=%o recv=%o",
                        c->sent, c->received);
    }
    *p++ = '\n';
    ngx_write_fd(ctx->file.fd, last, p - last);
}

7. 总结

本文从 nginx stream 模块的架构出发，详细讲解了核心数据结构、模块生命周期和开发接口，并通过一个完整的 stream 日志模块示例，展示了从配置解析、会话处理到日志写入的完整开发流程。stream 模块开发的关键要点包括：

理解会话模型：stream 模块基于连接而非请求，处理函数是 void 类型，通过 ngx_stream_core_run_phases 驱动处理链。
善用配置合并：通过 create_srv_conf 和 merge_srv_conf 实现灵活的 server 级别配置。
注册处理函数：在 postconfiguration 中将处理函数注册到 ngx_stream_content_phase 阶段。
连接生命周期管理：利用 connection->log->handler 注册连接关闭回调，实现断开事件记录。