第 39 章 日志（下）：五类 Sink、轮转与异步

本章导读

上一章写了自定义 Sink 的全部零件，本章组装整机：五类内置 Sink（控制台、文件、文本格式化、JSON 格式化、异步包装）覆盖绝大多数输出需求，配置器风格统一（ConfigInit + 定制 + 创建）。运维三件套——轮转（xrtLogFileRotate）、重开（xrtLogFileReopen，外部截断后恢复）、统计（xrtLogFileStats）让文件 Sink 适配长期运行。异步 Sink 是热路径的关键：提交线程只入队、专职线程写盘——日志不再拖慢业务。

39.1引入

日志上线的三个坎。坎一：日志要 JSON 格式进采集管道（ELK/Loki 吃结构化）——自定义 Sink 手写 JSON 拼装？字段转义、时间格式、嵌套错误链全是细节坑。坎二：日志文件无限增长——磁盘三个月被刷爆；要按大小轮转、要压缩归档、要在 logrotate 外部截断后还能恢复写入。坎三：高峰期每秒几万条日志，写盘的 IO 延迟直接加在请求耗时上——日志成了性能税。

三个坎对应三组答案：JSON Sink 开箱即用（字段自动进对象、错误链自动展开、时间毫秒戳）；文件族的 Rotate/Reopen/Stats（内置轮转语义 + 外部截断恢复）；异步 Sink 把写盘挪出提交路径（队列缓冲 + 专职线程刷盘）。它们共同的前提是第 38 章的地基——一切 Sink 都实现同一个写函数契约。

39.2概念

五类内置 Sink 一张表

配置器风格全库统一：ConfigInit 填默认 → 按需改字段 → 一步创建。JSON Sink 的输出形态（json 范例可见）：顶层 time（毫秒）/level/logger/message + fields 对象收纳结构化字段——字段名与类型原样保留，采集管道零解析成本。

文件运维三件套

轮转的触发分内置策略（按大小自动）与手动触发（xrtLogFileRotate 运维接口——按时间轮转由 cron 调它即可）；xrtLogFilePath 查询当前文件路径——轮转后路径变化，监控要知道当前写在哪。

异步 Sink：队列 + 专职线程

提交路径的真相：同步写文件 Sink 时，每条日志的 IO 延迟（毫秒级）直接加在业务调用上。异步 Sink 的结构：提交线程只做"格式化进队列"（毫秒级，无锁队列批量接口——第 21 章的 MPMC 正是它的底座）；专职线程批量出队写盘（批量 32 条的摊还效率）。代价与纪律：停机必须 Flush（xrtLogFlush——队列里的记录落盘完才算干净退出，丢的是崩溃前最后几百毫秒）；队列容量是背压阀（满时按策略丢弃或阻塞——丢弃计数进 Stats，"日志洪水牺牲日志"是正确取舍）。

格式化复用：TextWrite 与字段 printf

xrtLogTextWrite 族让自定义目的地复用文本格式化引擎（你的 Sink 只管"写去哪"，格式化交给库）；xrtLogFieldsPrintf 族是字段值的 printf 拼装（便捷层，第 38 章的选择规则照旧）。校验器（xrtLogTextConfigValidate / JsonConfigValidate / RecordValidate）在配置加载口把关——配置写错的格式串在启动时暴露，不是运行时炸雷。

39.3示例

完整程序：JSON 结构化输出

来自仓库范例 examples/logging/json/main.c——字段自动进对象、错误链展开：

 */

#include <stdio.h>
#include <string.h>

#include <xrt.h>



/* 输出回调：每个 JSON 分片一段（本例直写 stdout）。 */
static bool exampleLogWrite(xbytesview Data, ptr pUserData)
{
	FILE* pFile = (FILE*)pUserData;

	return fwrite(Data.Data, 1u, Data.Size, pFile) == Data.Size;
}



int main(void)
{
	xlogjsonconfig Config;
	xlogfield Fields[2];
	xlogrecord Record;

	/* 手工组装完整记录：时间、级别、来源、消息、两个结构化字段。 */
	memset(&Record, 0, sizeof(Record));
	Fields[0] = xrtLogFieldUInt(XRT_STR_LITERAL("request_id"), 42u);
	Fields[1] = xrtLogFieldBool(XRT_STR_LITERAL("cached"), false);
	Record.Time = xrtNow();
	Record.Level = XLOG_INFO;
	Record.Logger = XRT_STR_LITERAL("http");
	Record.Message = XRT_STR_LITERAL("request completed");
	Record.Fields = Fields;
	Record.FieldCount = 2u;

	/* 默认配置全字段输出；分片经回调直达目标。 */
	if (
		!xrtLogJsonConfigInit(&Config) ||
		!xrtLogJsonWrite(
			&Record,
			&Config,
			exampleLogWrite,
			stdout,
			NULL
		)
	) {
		fprintf(stderr, "log JSON failed: %s\n", xrtErrorMessage(xrtGetError()));
		return 1;
	}
	return 0;
}

刚才发生了什么。 ① 一条 LogFields 提交（整数字段 42、布尔字段 false）——JSON Sink 自动把它们收进 fields 对象：字段保类型直达（42 是数字不是 "42"、false 是布尔），采集管道的解析零成本。② time 是 公元 UTC 毫秒（第 3 章 xtime 口径）——机器可排序、人可格式化，格式选择权在消费端。③ 输出里 message 与 fields 分离——给人看的消息与给机器检索的字段各归各位，这是"结构化日志"的本义。对比第 38 章的文本 Sink（忽略字段只打消息）：同一个 Logger 挂两种 Sink，文本进控制台、JSON 进采集——双格式并行就是这个模型的价值。

完整程序：异步 Sink

来自 examples/logging/async/main.c——提交线程零阻塞、专职线程刷盘：

 */

#include <xrt.h>

#include <stdio.h>
#include <string.h>



/* 目标 Sink：直接输出原始消息（异步层不关心格式）。 */
static xlogresult exampleLogAsyncWrite(
	const xlogrecord* pRecord,
	ptr pUserData
)
{
	FILE* pFile = (FILE*)pUserData;

	if (
		fwrite(
			pRecord->Message.Data,
			1u,
			pRecord->Message.Size,
			pFile
		) != pRecord->Message.Size ||
		fputc('\n', pFile) == EOF
	) {
		return XLOG_RESULT_ERROR;
	}
	return XLOG_RESULT_WRITTEN;
}



/* 目标 Flush：由异步栅栏在工作线程上按序调用。 */
static bool exampleLogAsyncFlush(ptr pUserData)
{
	return fflush((FILE*)pUserData) == 0;
}



int main(void)
{
	xlogsinkconfig TargetConfig;
	xlogasyncconfig AsyncConfig;
	xlogrecord Record;
	xlogsink* pTarget;
	xlogsink* pAsync;

	/* 先造同步目标 Sink（stdout 直写）。 */
	memset(&TargetConfig, 0, sizeof(TargetConfig));
	TargetConfig.Name = XRT_STR_LITERAL("stdout");
	TargetConfig.Level = XLOG_TRACE;
	TargetConfig.Write = exampleLogAsyncWrite;
	TargetConfig.Flush = exampleLogAsyncFlush;
	TargetConfig.UserData = stdout;
	pTarget = xrtLogSinkCreate(&TargetConfig);
	if ( pTarget == NULL ) {
		return 1;
	}

	/* 配置背压参数后包装：Async 持有目标引用，本地引用可释放。 */
	if ( !xrtLogAsyncConfigInit(&AsyncConfig) ) {
		xrtLogSinkFree(pTarget);
		return 2;
	}
	AsyncConfig.Capacity = 256u;
	AsyncConfig.ByteLimit = 2u * 1024u * 1024u;
	pAsync = xrtLogAsync(pTarget, &AsyncConfig);
	xrtLogSinkFree(pTarget);
	if ( pAsync == NULL ) {
		return 3;
	}

	/* 提交 = 入队即返回；业务线程不等待写盘。 */
	memset(&Record, 0, sizeof(Record));
	Record.Level = XLOG_INFO;
	Record.Message = XRT_STR_LITERAL("asynchronous sink ready");
	if ( xrtLogSinkSubmit(pAsync, &Record) != XLOG_RESULT_WRITTEN ) {
		xrtLogSinkFree(pAsync);
		return 4;
	}

	/* 栅栏冲刷：返回即保证该条已真实写出。 */
	if ( !xrtLogSinkFlush(pAsync) ) {
		xrtLogSinkFree(pAsync);
		return 5;
	}
	xrtLogSinkFree(pAsync);
	return 0;
}

刚才发生了什么。 ① 异步 Sink 包装一个真实目的地（文件）——对 Logger 而言它就是普通 Sink，提交接口不变。② 主线程提交记录只入队（asynchronous sink ready 立即打印——提交没有被写盘阻塞）；专职线程在后台批量消费。③ 停机路径的 Flush：队列清空、文件句柄关闭——干净退出。把 sink_tour 范例留作运维接口的完全体阅读：它把 Ref/Name/Level/SetLevel/Count/Stats/Detach 族与文件专用族（Path/Rotate/Reopen/Stats）全部走了一遍，九个 ok 对应九组接口。

39.4契约

• 配置器统一：ConfigInit → 定制 → 创建；校验器在配置口把关。
• JSON 形态：time 毫秒/level/logger/message + fields 对象——字段保类型、时间机器序。
• 轮转语义：Rotate 改名归档开新文件；按大小自动 + 手动触发；Path 查当前路径。
• 截断恢复：外部 move/截断后 Reopen 重开——logrotate 协作的内置支持。
• 异步纪律：停机 Flush；队列容量是背压阀，丢弃计数进 Stats；提交毫秒级、刷盘批量摊还。
• 格式化复用：自定义目的地用 TextWrite 族；给机器的字段仍走 LogFields。

从示例到工程：Sink 装配的三个模式

开发环境：一个控制台 Sink（文本格式、DEBUG 门槛）——终端即时可读，五分钟装配。生产标准：JSON 文件 Sink（INFO 门槛、按大小轮转）+ 异步包装（队列背压）+ 控制台 ERROR Sink（容器环境的 stderr 告警通道）——三个 Sink 各司其职，一次装配。采集管道：JSON 格式直连 stdout（容器 stdout 采集）或轮转文件 + Filebeat 类代理——格式已在 Sink 层统一，管道零适配。三个模式的共同形状：装配代码集中在启动函数（第 38 章纪律），配置来自第 32 章 JSON 配置文件（Sink 类型、路径、级别全部可配）——"日志系统的日志配置"是配置驱动装配的标准练习题。

一个运维观：日志是数据流

把日志当成"打印"是开发视角；上线后它是数据流——有源头（业务代码）、有管道（Sink 与队列）、有归宿（文件/采集系统）、有容量（磁盘与队列）、有生命周期（轮转与归档）。用数据流的视角做运维决策：容量规划看 Stats 的写入速率 × 保留周期；洪水防护靠队列丢弃策略 + 级别动态下调（SetLevel 运行时可调）；故障恢复验证 Reopen（演练外部截断）；质量监控对账"提交量 vs 落盘量"（异步丢弃在容忍范围）。这个视角也解释了第 36 章管线思想的再次适用——日志系统本身就是"记录 → 格式化 → 队列 → 落盘"的管线，你在卷四学的组装能力在这里直接上岗。

常见错误

/* 停机 */
xrtLogSinkFree(pAsyncSink);   /* 直接释放——队列内容丢弃 */

/* 停机：先 Flush 队列再释放 */
xrtLogFlush(pLogger);          /* 或异步 Sink 专属 Flush 接口 */
xrtLogSinkFree(pAsyncSink);

void handle_request(void)
{
	xlogsink* pSink = create_file_sink("app.log");   /* 每请求建一个 */
	xrtLogAttach(gLog, pSink);
	xrtLog(gLog, XLOG_INFO, ...);
	xrtLogSinkFree(pSink);                            /* 拆了建、建了拆 */
}

/* 启动时装配一次，全局复用（第 38 章"装配集中"纪律） */
static xlogsink* gFileSink;
void init_logging(void) { gFileSink = create_file_sink("app.log"); xrtLogAttach(gLog, gFileSink); }
void handle_request(void) { xrtLog(gLog, XLOG_INFO, ...); }   /* 只提交，不装配 */

编程练习

小结