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

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

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

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

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

更容易维护的结构是：多个 Handler 视图共享一个提交核心。Handler 视图只保存 WithAttrs 和 WithGroup 形成的不可变派生状态；提交核心拥有有界队列、唯一批量工作协程和 Sink。所有网络提交都发生在这个工作协程里，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。

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

下面的代码把传输抽象成 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 错误返回给业务代码。

测试时至少覆盖这些边界

• 达到 batchSize 时立即提交，低流量时由 flushEvery 提交。
• 并发调用 Handle 不出现数据竞争，关闭后不发生向已关闭 channel 发送。
• WithAttrs 和 WithGroup 的交错顺序保持正确，派生 Handler 不修改原 Handler。
• LogValuer 在入队前解析，可变业务对象稍后变化不会改写已入队事件。
• Sink 失败能从指标和 Close 看见，失败批次的重试或丢弃策略符合业务约定。
• 用 testing/slogtest 配合内存 Sink 检查 Handler 对 Record、Group 和 Attr 的处理。

相关问题

为什么不直接把 slog.Record 放进 channel？

至少要先调用 Clone，避免共享内部状态；如果 Attr 含 LogValuer 或可变引用，还应在调用协程建立序列化快照，否则后台读取时值可能已经变化。

队列满时返回 error 可以防止丢日志吗？

不一定。常用的 slog.Logger 日志方法不会把 Handler 错误返回给业务调用方，所以非阻塞丢弃还必须配套计数器、告警或同步兜底。

batchSize 越大越好吗？

不是。大批次能减少请求次数，却会增加内存占用和低流量等待时间。应同时观察批次填充率、p95 提交延迟、队列深度和失败重试成本。

Handler 里能因为 context 取消而跳过日志吗？

不建议。官方接口说明上下文主要用于向 Handler 传递信息，取消本身不应阻止日志记录；如果业务允许丢弃，应通过明确的采样或队列策略表达。

参考资料

• Go log/slog：https://pkg.go.dev/log/slog
• Go slog Handler 指南：https://go.dev/s/slog-handler-guide
• Go 官方博客 Structured Logging with slog：https://go.dev/blog/slog