zap WriteSyncer与Sink体系

本文目录

2.1 AddSync：类型适配
2.2 Lock：并发安全包装
2.3 multiWriteSyncer：多路
3.1 惰性初始化（112-133）
3.2 Write 的防撕裂处理（148-152）
3.3 flushLoop 与 Stop（170-220）
3.4 Clock 抽象
4.1 注册表结构
4.2 newSink 的路由逻辑（93-117）
4.3 file 工厂的校验（130-159）
4.4 RegisterSink 的合法性检查（161-180）
4.5 Open：串起一切（writer.go:50-98）
6.1 loggerWriter：最小的桥
6.2 caller 深度补偿
6.3 全局 logger 的锁

1. WriteSyncer 家族全景 -------------------

                     WriteSyncer 接口（io.Writer + Sync）
                                  │
     ┌──────────────┬─────────────┼──────────────┬────────────────┐
     │              │             │              │                │
 *os.File     writerWrapper   lockedWriteSyncer  multiWriteSyncer  BufferedWriteSyncer
 (天然实现)    (AddSync 包装    (Lock 加锁)       (Multi 多路)     (攒批+定时刷)
               无Sync的Writer)

基础组合子（write_syncer.go）

2.1 AddSync：类型适配

// write_syncer.go:40-47
func AddSync(w io.Writer) WriteSyncer {
    switch w := w.(type) {
    case WriteSyncer:
        return w                      // 已实现 → 原样（如 *os.File）
    default:
        return writerWrapper{w}       // 否则补一个 no-op Sync
    }
}

2.2 Lock：并发安全包装

// write_syncer.go:56-62
func Lock(ws WriteSyncer) WriteSyncer {
    if _, ok := ws.(*lockedWriteSyncer); ok {
        return ws                     // 幂等：不会套两层锁
    }
    return &lockedWriteSyncer{ws: ws}
}
// Write/Sync 都在 mutex 内（64-76）

为什么需要锁？ os.File 的单次 Write 是原子的，但"Write + Sync"组合不原子；更重要的是自定义 Writer（如 bufio.Writer、你的 Kafka sink）大多非并发安全。zap 的规则：New(core) 默认给 errorOutput 加 Lock（logger.go:75），Open 返回值自动 Lock（writer.go:97）；手工传给 NewCore 的 WriteSyncer 自己负责。

2.3 multiWriteSyncer：多路

// write_syncer.go:90-122
func NewMultiWriteSyncer(ws ...WriteSyncer) WriteSyncer {
    if len(ws) == 1) { return ws[0] }       // 单个不包
    return multiWriteSyncer(ws)
}
// Write：全部写；返回最小 n；错误 multierr 合并（101-114，模仿 io.MultiWriter）
// Sync：全部刷；错误合并（116-122）

BufferedWriteSyncer：攒批写入

buffered_write_syncer.go，结构（77-110）：

type BufferedWriteSyncer struct {
    WS            WriteSyncer  // 底层目标（必填）
    Size          int          // 缓冲上限，默认 256KB
    FlushInterval time.Duration // 定时刷间隔，默认 30s
    Clock         Clock        // 时钟（测试注入 ticker）

    mu          sync.Mutex
    initialized bool
    stopped     bool
    writer      *bufio.Writer
    ticker      *time.Ticker
    stop, done  chan struct{}
}

3.1 惰性初始化（112-133）

func (s *BufferedWriteSyncer) initialize() {
    // 填默认值 → 创建 bufio.Writer → 起后台 flushLoop goroutine
    go s.flushLoop()
}
// Write 里第一次调用时才 initialize（137-155）——没写过日志就没有 goroutine 开销

3.2 Write 的防撕裂处理（148-152）

// 当前写入放不进缓冲 且 缓冲非空 → 先手动 Flush
// （bufio 对空缓冲的大写入不会拆分，这里保证单条日志不被截成两次底层写）
if len(bs) > s.writer.Available() && s.writer.Buffered() > 0 {
    if err := s.writer.Flush(); err != nil { return 0, err }
}
return s.writer.Write(bs)

3.3 flushLoop 与 Stop（170-220）

func (s *BufferedWriteSyncer) flushLoop() {
    defer close(s.done)
    for {
        select {
        case <-s.ticker.C: _ = s.Sync()     // 定时刷（错误先吞，bufio 会记）
        case <-s.stop:     return
        }
    }
}

func (s *BufferedWriteSyncer) Stop() (err error) {
    stopped := func() bool {                 // 临界区只做标记和关信号
        s.mu.Lock(); defer s.mu.Unlock()
        if !s.initialized || s.stopped { return false }
        s.stopped = true
        s.ticker.Stop()
        close(s.stop)
        return true
    }()
    if !stopped { return }
    <-s.done      // ★ 锁外等待 goroutine 退出（锁内等会死锁，见 issue #1428 注释）
    return s.Sync()  // 最后刷一次
}

设计细节：Stop 的等待放锁外——flushLoop 的收尾需要拿锁，锁内等它会死锁。这是"锁内做最少事"的经典示范。

3.4 Clock 抽象

zapcore/clock.go：Clock 接口 = Now() + NewTicker()，DefaultClock 是系统实现——为了测试能造"手动前进时间"的 ticker（clock_test.go）。

Sink：URL 到 WriteSyncer 的工厂体系

4.1 注册表结构

sink.go:58-72：

type sinkRegistry struct {
    mu        sync.Mutex
    factories map[string]func(*url.URL) (Sink, error)  // scheme → 工厂
    openFile  func(string, int, os.FileMode) (*os.File, error)  // 可替换的 os.OpenFile（测试）
}

// 初始化时注册 file scheme（70 行）
_ = sr.RegisterSink(schemeFile, sr.newFileSinkFromURL)

Sink = WriteSyncer + io.Closer（41-44）——多一个 Close 生命周期。

4.2 newSink 的路由逻辑（93-117）

func (sr *sinkRegistry) newSink(rawURL string) (Sink, error) {
    // ① Windows 兼容：绝对路径直接按文件开（c:\log.txt 会被 url.Parse 误判 scheme="c"）
    if filepath.IsAbs(rawURL) {
        return sr.newFileSinkFromPath(rawURL)
    }
    // ② 解析 URL；无 scheme 补 "file"
    u, err := url.Parse(rawURL)
    if u.Scheme == "" { u.Scheme = schemeFile }

    // ③ 查注册表调工厂
    factory, ok := sr.factories[u.Scheme]
    if !ok { return nil, &errSinkNotFound{u.Scheme} }
    return factory(u)
}

4.3 file 工厂的校验（130-159）

newFileSinkFromURL 拒绝 file URL 带 user/fragment/query/port/非 localhost host；newFileSinkFromPath 特判 "stdout"/"stderr"，其余 O_WRONLY|O_APPEND|O_CREATE, 0666。

4.4 RegisterSink 的合法性检查（161-180）

normalizeScheme：必须小写字母开头，后续 [a-z0-9.+-]（RFC 3986 3.1 节）——先小写化再校验，所以 scheme 大小写不敏感。

4.5 Open：串起一切（writer.go:50-98）

func Open(paths ...string) (zapcore.WriteSyncer, func(), error) {
    writers, closeAll, err := open(paths)     // 逐个 newSink；任一失败关闭已开的
    writer := CombineWriteSyncers(writers...)
    return writer, closeAll, nil
}

func CombineWriteSyncers(writers ...zapcore.WriteSyncer) zapcore.WriteSyncer {
    if len(writers) == 0 { return zapcore.AddSync(io.Discard) }  // 空目标 = 丢弃
    return zapcore.Lock(zapcore.NewMultiWriteSyncer(writers...))
}

注意 closeAll 闭包捕获 closers——Config.Build 拿到后只用于出错回滚（config.go:315-326）；正常路径的文件句柄生命周期跟随进程（这也是为什么改 OutputPaths 要重建 logger）。

编码器注册表（zap/encoder.go）

平行的另一张表（encoder.go:34-43）：

var _encoderNameToConstructor = map[string]func(zapcore.EncoderConfig) (zapcore.Encoder, error){
    "console": → NewConsoleEncoder,
    "json":    → NewJSONEncoder,
}

func RegisterEncoder(name string, ctor ...) error   // 51-62：重名报错
func newEncoder(name, cfg) (Encoder, error)          // 64-79：
    // ★ TimeKey 非空但 EncodeTime == nil → "missing EncodeTime" 错误的出处

Config.Encoding 字符串最终就是查这张表。

标准库桥接与全局 logger（global.go）

6.1 loggerWriter：最小的桥

// global.go:161-169
type loggerWriter struct {
    logFunc func(msg string, fields ...Field)   // 绑定了某个级别方法
}
func (l *loggerWriter) Write(p []byte) (int, error) {
    p = bytes.TrimSpace(p)      // 去掉 log 包加的换行
    l.logFunc(string(p))
    return len(p), nil
}

6.2 caller 深度补偿

// global.go:33-35
_stdLogDefaultDepth = 1   // log.Output 的内部栈深
_loggerWriterDepth  = 2   // loggerWriter.Write + 被绑定的级别方法

// NewStdLog（78-82）：l.WithOptions(AddCallerSkip(3)) 再绑定 logger.Info
// RedirectStdLogAt（123-139）：记下原 flags/prefix → SetFlags(0)+SetPrefix("")
//   → log.SetOutput(&loggerWriter{logFunc}) → 返回还原闭包

6.3 全局 logger 的锁

// global.go:40-73
var (
    _globalMu sync.RWMutex        // 写少读多 → RWMutex
    _globalL  = NewNop()          // 默认 Nop！
    _globalS  = _globalL.Sugar()
)
L()/S()：RLock 读 → 返回副本指针
ReplaceGlobals(l)：Lock 写 L + 重新 Sugar S；返回还原函数（递归调用自己，restore 旧值）

写入路径完整时序（串前两篇）

ce.Write(fields)
 └─ ioCore.Write (core.go:94)
     ├─ buf = enc.EncodeEntry(ent, fields)     [13 篇]
     ├─ c.out.Write(buf.Bytes())  ─────────────▶ 本篇：
     │     out 可能是：
     │       lockedWriteSyncer(multi(stderr, file))    ← Open 的产物
     │       BufferedWriteSyncer(→ bufio → lumberjack) ← 手工组装
     │       自定义 Sink(kafka/tcp/...)                ← RegisterSink
     ├─ buf.Free()                              [buffer 池]
     └─ Fatal 级 → c.Sync()