当前位置：首页 > 文章列表 > Golang > Go教程 > slog 自定义 Handler 怎样批量提交结构化日志

slog 自定义 Handler 怎样批量提交结构化日志

来源：17golang原创 2026-10-09 15:45:29 0浏览收藏

slog.Handler 要批量提交结构化日志，推荐把它设计成“同步快照、并发入队、单协程聚合”：Handle 在调用协程里把 slog.Record 转成不可变事件，然后写入有界队列；后台工作协程按条数或时间窗口组成批次，再调用 Sink.WriteBatch。这样既能减少网络往返，也能把并发、背压和关闭刷新边界讲清楚。

最小结论

Handler 的方法可能被并发调用，共享状态必须自行同步。
WithAttrs、WithGroup 必须返回新 Handler，不能修改原接收者。
异步处理前先建立事件快照，避免稍后读取已经变化的 Record 或业务对象。
队列满时必须明确选择阻塞、丢弃或同步回退；本文示例选择背压。
Close 要停止入口、排空队列并返回异步提交错误。

官方文档：https://pkg.go.dev/log/slog

接口目标：Handle 只做快照和入队

slog.Logger 在日志级别通过 Enabled 检查后构造 Record，再调用 Handler 的 Handle。如果每次 Handle 都直接请求远端接口，请求线程会承担完整网络延迟；如果为每条日志启动 goroutine，又会失去并发上限和关闭边界。

更容易维护的结构是：多个 Handler 视图共享一个提交核心。Handler 视图只保存 WithAttrs 和 WithGroup 形成的不可变派生状态；提交核心拥有有界队列、唯一批量工作协程和 Sink。所有网络提交都发生在这个工作协程里，Sink 因此不需要处理多个批次同时写入。

slog Logger、派生 Handler、共享 core、事件队列、批次工作协程和 Sink 的静态所有权结构图 — 图1：批量 Handler 的派生状态与共享提交核心结构图，不是运行截图。

调用方需求：吞吐、尾延迟和可靠性要分开决定

需求	对应设计	需要接受的代价
减少网络请求	达到 batchSize 后一次提交	低流量日志会等待成批
限制等待时间	flushEvery 到期时提交残留	批次可能小于 batchSize
不静默丢日志	有界队列满时阻塞 Handle	日志高峰会反压业务协程
业务协程绝不阻塞	非阻塞入队并统计丢弃量	必须接受并监控日志损失
进程退出前尽量提交	Close 关闭入口并排空队列	关闭过程可能等待远端 Sink

没有一种队列策略适合所有业务。审计日志通常宁愿阻塞或落盘，也不能静默丢弃；高频调试日志则可能更适合采样或丢弃。策略应由日志用途决定，而不是藏在 Handler 内部成为偶然行为。

参数设计：先生成事件快照，再交给工作协程

slog.Record.Clone 会生成不共享内部状态的副本，但异步 Handler 还要考虑 Attr 中的 LogValuer 或可变对象。本文示例在 Handle 内完成 Resolve 和 JSON 安全转换，把值固定成字符串、数字、布尔值或普通 JSON 数据，再入队。这样工作协程只处理自己的 Event，不会晚一步读取业务对象。

分组顺序也不能被压扁成“所有 attrs 加所有 groups”。例如 WithGroup("request").With("id", 7).WithGroup("db") 中，id 属于 request，不属于 request.db。因此代码使用有序的 scopeItem 保存每次 WithAttrs 或 WithGroup。

slog Record Clone、分组状态、属性解析、Event 字段、批次缓冲与异步错误的静态数据关系图 — 图2：slog.Record 到不可变 Event 快照的字段关系图，不是运行截图。

完整示例：按条数或周期批量提交

下面的代码把传输抽象成 Sink。示例选择可靠性优先的背压：队列已满时，Handle 等待工作协程腾出空间。异步 Sink 错误记录为首个错误，并在 Close 时返回；示例不会自动重试失败批次，生产环境可在 Sink 层增加有限重试或磁盘缓冲。

package batchslog

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"
)

var ErrClosed = errors.New("batch slog handler is closed")

// Event 是提交给远端的不可变结构化事件。
type Event struct {
	Time    time.Time      `json:"time"`
	Level   string         `json:"level"`
	Message string         `json:"message"`
	Fields  map[string]any `json:"fields"`
}

// Sink 负责同步消费一个批次；返回前不得继续持有 batch 切片。
type Sink interface {
	WriteBatch(ctx context.Context, batch []Event) error
}

type scopeItem struct {
	group string
	attrs []slog.Attr
}

type core struct {
	sink       Sink
	queue      chan Event
	done       chan struct{}
	batchSize  int
	flushEvery time.Duration

	mu     sync.RWMutex
	closed bool
	stop   sync.Once
	errMu  sync.Mutex
	err    error
}

type BatchHandler struct {
	core     *core
	minLevel slog.Level
	state    []scopeItem
}

// New 创建共享提交核心，并启动唯一的批量工作协程。
func New(sink Sink, queueSize, batchSize int, flushEvery time.Duration) *BatchHandler {
	c := &core{
		sink:       sink,
		queue:      make(chan Event, queueSize),
		done:       make(chan struct{}),
		batchSize:  batchSize,
		flushEvery: flushEvery,
	}
	go c.run()
	return &BatchHandler{core: c, minLevel: slog.LevelInfo}
}

func (h *BatchHandler) Enabled(_ context.Context, level slog.Level) bool {
	// 尽早过滤低级别日志，避免构造字段快照。
	return level >= h.minLevel
}

func (h *BatchHandler) Handle(_ context.Context, r slog.Record) error {
	// 在调用协程中解析所有值，工作协程不再访问业务对象。
	e := makeEvent(r.Clone(), h.state)

	h.core.mu.RLock()
	defer h.core.mu.RUnlock()
	if h.core.closed {
		return ErrClosed
	}
	// 有界队列满时显式背压；不要用 context 取消来跳过日志。
	h.core.queue = c.batchSize {
				flush()
			}
		case

错误处理：Handle 成功不等于远端已经写入

异步 Handler 的 Handle 最多只能说明“事件已快照并进入本地队列”。真正的远端错误发生在另一个 goroutine，无法原样同步返回给当前日志调用。示例把首个异步错误保存到 core，并由 Close 返回；实际服务还应把提交失败数、队列深度、批次大小和提交耗时暴露成指标。

示例在 Sink 失败后丢弃该内存批次，因此属于有限可靠性的骨架。如果日志必须持久保存，可在 Sink 中增加有上限的退避重试，或先写本地 WAL 再确认消费。不要无限重试同一失败批次，否则队列会被一个永久错误堵死。

兼容策略：不要破坏 WithAttrs 与 WithGroup

Go 官方 Handler 指南强调，WithAttrs 和 WithGroup 要返回新的 Handler，原 Handler 保持不变，而且两种调用的顺序必须被保留。本文让派生视图共享 core，但复制 state 切片，因此新 Logger 的分组与固定属性不会污染旧 Logger。

示例把分组键扁平化为 request.db.duration。如果远端支持嵌套 JSON，可以把 makeEvent 改成构造嵌套 map；只要不同分组序列最终产生不同键空间，并保持空组、重复键和 Group 的约定一致即可。

调用示例与关闭顺序

// HTTPBatchSink 是业务提供的远端批量提交实现。
sink := NewHTTPBatchSink("https://logs.example.internal/v1/batch")
handler := batchslog.New(sink, 2048, 100, 2*time.Second)
logger := slog.New(handler)

// 固定属性与分组会保存在派生 Handler 中，不修改原 logger。
requestLog := logger.WithGroup("request").With(
	slog.String("service", "checkout"),
)
requestLog.Info("payment accepted",
	slog.String("order_id", "A-1024"),
	slog.Int64("cost_ms", 38),
)

// 先停止产生新日志的 goroutine，再关闭并检查最后一次批量提交。
if err := handler.Close(); err != nil {
	// 关闭阶段只能使用独立的兜底输出，不能再次写入已关闭的 handler。
	fmt.Fprintf(os.Stderr, "flush structured logs: %v\n", err)
}

如果派生 Logger 仍可能在其他 goroutine 中使用，应先停止这些生产者，再调用根 Handler 的 Close。否则关闭会与新日志竞争，调用方只能收到 ErrClosed，而 slog.Logger 的便捷日志方法不会把 Handler 错误返回给业务代码。