首页 > 服务器    日期:2026-07-20 / 浏览

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 模块类似,但更简洁:

  1. 配置解析阶段:读取 stream {} 块中的指令,调用模块的 create_main_confcreate_srv_confpreconfiguration 回调。
  2. 初始化阶段init_main_confinit_worker 回调。
  3. 会话处理阶段:当新连接到达时,调用模块的 handler 回调处理会话。
  4. 销毁阶段exit_masterexit_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 模块开发的关键要点包括:

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

掌握这些知识后,你可以开发各种自定义 Stream 模块,如协议解析器、流量整形器、自定义负载均衡策略等。

觉得上面的内容有用吗?快来点个赞吧!

点赞() 我要打赏

温馨提示 : 本站内容来自会员投稿以及互联网,所有源码及教程均为作者总结编辑,请大家在使用过程中提前做好备份,以免发生无法预知的错误,源码类教程请勿直接用于生产环境!

 可能感兴趣的文章