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注册连接关闭回调,实现断开事件记录。
掌握这些知识后,你可以开发各种自定义 Stream 模块,如协议解析器、流量整形器、自定义负载均衡策略等。













