go-zero 源码分析 12：时间与并发工具箱

这篇文章将学习 go-zero 的一些底层组件，这些工具本身不参与业务逻辑，也不直接暴露给最终用户。它们构成了框架的"标准库"——如果上层建筑是房屋的梁柱，这些工具就是砌成梁柱的砖石。它们分布在 core/collection/、core/syncx/、core/executors/、core/fx/、core/mr/ 和 core/threading/ 六个包中。我们不会逐个罗列所有类型，而是从上层模块的依赖关系出发，沿着"时间工具 → 并发控制 → 执行器 → 高级抽象"这条线索来组织它们。

时间驱动：从 TimingWheel 到 RollingWindow

在 go-zero 的工具箱里，有两类与时间相关的核心数据结构：时间轮处理"在未来的某个时刻做什么"，滚动窗口处理"过去一段时间内发生了什么"。前者是延迟任务调度器，后者是滑动统计算子。

TimingWheel：用空间换时间的延迟调度器

我们先看一个具体的问题：core/collection/cache.go 中的 Cache 组件需要一种机制，能在每个缓存项过期时自动将其删除。最直觉的实现方式是给每个缓存项分配一个 time.Timer。但当缓存项数量达到数十万时，操作系统需要维护数十万个活跃的定时器——每个定时器都是一个内核资源，这会迅速拖垮性能。

时间轮（Timing Wheel）是这个问题的经典解法。它的核心思想很简单：用一个环形数组（槽）和一个循环 ticker 替代独立的定时器。

go-zero 的时间轮实现在 core/collection/timingwheel.go。先看它的结构：

// core/collection/timingwheel.go
type TimingWheel struct {
    interval      time.Duration   // tick 间隔
    ticker        timex.Ticker    // 驱动 tick 的定时器
    slots         []*list.List    // 环形槽数组，每个槽存一个任务链表
    timers        *SafeMap        // key → positionEntry，快速查找任务位置
    tickedPos     int             // 当前 tick 所在的槽位
    numSlots      int             // 槽总数
    execute       Execute         // 任务到期时的回调
    setChannel    chan timingEntry  // 添加/更新任务的 channel
    moveChannel   chan baseEntry    // 移动任务的 channel
    removeChannel chan any          // 删除任务的 channel
    drainChannel  chan func(key, value any) // 导出所有任务的 channel
    stopChannel   chan lang.PlaceholderType
}

与经典的 Hashed-and-Hierarchical Timing Wheel 不同，go-zero 采用了一个单层设计——没有多层级的轮。那它如何支持任意时长的延迟呢？答案是引入了 circle（圈数）和 diff（槽内偏移）两个字段：

type timingEntry struct {
    baseEntry                 // delay + key
    value   any               // 任务数据
    circle  int               // 还需要绕几圈
    diff    int               // 槽内微调偏移
    removed bool              // 是否被删除了
}

当一个任务被设置时，系统计算它距离当前 tick 位置有多少个 interval 步长。步长的一部分用来确定目标槽位，另一部分用来确定圈数：

func (tw *TimingWheel) getPositionAndCircle(d time.Duration) (pos, circle int) {
    steps := int(d / tw.interval)
    pos = (tw.tickedPos + steps) % tw.numSlots
    circle = (steps - 1) / tw.numSlots
    return
}

例如，一个时间轮有 60 个槽，tick 间隔为 1 秒。当前指针在槽 0，设置一个 5 秒延迟的任务时：steps = 5，pos = 5，circle = (5-1)/60 = 0——任务放在第 5 个槽，圈数为 0。但如果设置一个 125 秒延迟的任务：steps = 125，pos = (0+125)%60 = 5，circle = (125-1)/60 = 2——任务也放在第 5 个槽，但圈数标记为 2，意味着指针需要再绕两圈到这个槽时，任务才算真正到期。

每次 tick 触发 onTick，指针前进一个槽位，然后扫描该槽的任务链表：

func (tw *TimingWheel) scanAndRunTasks(l *list.List) {
    var tasks []timingTask
    for e := l.Front(); e != nil; {
        task := e.Value.(*timingEntry)
        if task.removed {        // 已删除，从链表移除
            next := e.Next()
            l.Remove(e)
            e = next
            continue
        } else if task.circle > 0 {  // 还没绕够圈数，递减
            task.circle--
            e = e.Next()
            continue
        } else if task.diff > 0 {    // 需要微调偏移，移到目标槽
            next := e.Next()
            l.Remove(e)
            pos := (tw.tickedPos + task.diff) % tw.numSlots
            tw.slots[pos].PushBack(task)
            tw.setTimerPosition(pos, task)
            task.diff = 0
            e = next
            continue
        }
        tasks = append(tasks, timingTask{key: task.key, value: task.value})
        // 从链表删除并清理索引
        next := e.Next()
        l.Remove(e)
        tw.timers.Del(task.key)
        e = next
    }
    tw.runTasks(tasks)  // 在新 goroutine 中执行到期任务
}

扫描逻辑有四个分支，优先级从高到低：

1. 已删除：从链表中移除即可
2. 圈数大于 0：圈数减一，留在槽中等待下一轮
3. diff 大于 0：任务需要微调到后续的某个槽位（MoveTimer 后的残余偏移）
4. 真正到期：收集到 tasks 列表中，最后在新 goroutine 中批量执行

diff 字段值得单独说明。当 MoveTimer 移动一个任务时，如果新的延迟比当前 tick 间隔还短，就直接执行；否则计算新的目标位置。但有时新位置正好等于当前槽位——这时不能直接执行（延迟还没到），diff 就起到了"槽内偏移"的作用，让任务在下一轮扫描时移动到正确的槽位。

所有对时间轮的操作（SetTimer、MoveTimer、RemoveTimer）都通过 channel 发给后台 goroutine run() 来串行处理，保证了并发安全。这种 actor 模式在 go-zero 中很常见——用一个 goroutine 作为事件循环，外部通过 channel 通信。

值得注意的是 MoveTimer 的实现细节：

func (tw *TimingWheel) moveTask(task baseEntry) {
    val, ok := tw.timers.Get(task.key)
    if !ok {
        return
    }
    timer := val.(*positionEntry)
    if task.delay < tw.interval {
        // 延迟小于 tick 间隔，直接执行
        threading.GoSafe(func() {
            tw.execute(timer.item.key, timer.item.value)
        })
        return
    }
    pos, circle := tw.getPositionAndCircle(task.delay)
    if pos >= timer.pos {
        timer.item.circle = circle
        timer.item.diff = pos - timer.pos
    } else if circle > 0 {
        circle--
        timer.item.circle = circle
        timer.item.diff = tw.numSlots + pos - timer.pos
    } else {
        // 新位置在当前之前且圈数为 0，需要用新节点
        timer.item.removed = true
        newItem := &timingEntry{baseEntry: task, value: timer.item.value}
        tw.slots[pos].PushBack(newItem)
        tw.setTimerPosition(pos, newItem)
    }
}

当一个任务的新槽位在当前槽位之后（pos >= timer.pos），只需要更新圈数和 diff 偏移，任务还在原槽位中——下次扫描时 diff > 0 分支会把它移走。当新槽位在之前且圈数大于 0 时，圈数减一，偏移绕一圈计算。而当新位置在前面且圈数已经为 0 时，说明这个延迟比原来的还短——这就尴尬了，因为任务已经经过了这个槽位。此时采用重建节点的方式：标记旧节点已删除，在新槽位创建一个新节点。

Cache：时间轮的典型消费者

理解了时间轮后，再看 Cache 组件就很简单了。Cache 组合了三个核心工具：

type Cache struct {
    data           map[string]any        // 数据存储
    expire         time.Duration          // 默认过期时间
    timingWheel    *TimingWheel           // 过期调度
    lruCache       lru                    // LRU 淘汰（可选）
    barrier        syncx.SingleFlight     // 防击穿
    unstableExpiry mathx.Unstable         // 随机过期偏移
}

Cache 创建时初始化了一个 300 槽、每秒 tick 一次的时间轮：

timingWheel, _ := NewTimingWheel(time.Second, slots, func(k, v any) {
    key, ok := k.(string)
    if !ok {
        return
    }
    cache.Del(key)
})

SetWithExpire 方法将缓存项放入 data map，同时在时间轮中注册一个到期回调。注意 unstableExpiry.AroundDuration(expire) 这一步——它给过期时间增加了一个 ±5% 的随机抖动。这个设计的目的是防止缓存雪崩：如果大量缓存在同一时刻过期，回源请求会瞬间压垮数据库。通过随机化过期时间，缓存失效被分散到一个时间窗口内。

Cache 还支持可选的 LRU 淘汰（通过 WithLimit 选项）。LRU 使用 container/list 实现——新访问的 key 移到链表头部，超出容量时从尾部淘汰。淘汰时同时清理数据 map 和时间轮中的定时器。

Take 方法是 Cache 最复杂的操作——它融合了 双重检查 + SingleFlight：

func (c *Cache) Take(key string, fetch func() (any, error)) (any, error) {
    if val, ok := c.doGet(key); ok {    // 第一次检查
        return val, nil
    }
    var fresh bool
    val, err := c.barrier.Do(key, func() (any, error) {
        if val, ok := c.doGet(key); ok { // 第二次检查（barrier 内部）
            return val, nil
        }
        v, e := fetch()
        // ...
        fresh = true
        c.Set(key, v)
        return v, nil
    })
    // ...
}

第一次 doGet 是快速路径——大多数情况下缓存命中，直接返回。barrier.Do 确保对于同一个 key，只有第一个 goroutine 会执行 fetch 回源，后续 goroutine 等待第一个的结果。而进入 barrier 后的第二次 doGet 是双重检查——可能在等待前一个 barrier.Do 调用时，缓存已经被填入了。

RollingWindow：时间维度的滑动统计算子

如果 TimingWheel 回答的是"未来做什么"，RollingWindow 回答的就是"过去发生了什么"。

我们在熔断器、降载器等组件中已经看到滚动窗口的身影。回到 core/collection/rollingwindow.go，它的结构很紧凑：

type RollingWindow[T Numerical, B BucketInterface[T]] struct {
    lock          sync.RWMutex
    size          int              // 桶个数
    win           *window[T, B]    // 桶数组
    interval      time.Duration    // 每个桶的时间跨度
    offset        int              // 当前桶在数组中的位置
    ignoreCurrent bool             // Reduce 时是否忽略当前桶
    lastTime      time.Duration    // 当前桶的起始时间
}

窗口 = 桶数组 + 时间间隔 + 偏移量。例如，一个 size=10、interval=1s 的滚动窗口代表"过去 10 秒的数据"，每一秒的数据积累在一个桶中。

写入操作很简单——每次 Add(v) 先更新偏移（将过期的桶清零），再向当前桶写入：

func (rw *RollingWindow[T, B]) Add(v T) {
    rw.lock.Lock()
    defer rw.lock.Unlock()
    rw.updateOffset()
    rw.win.add(rw.offset, v)
}

偏移更新的核心逻辑在 updateOffset：

func (rw *RollingWindow[T, B]) updateOffset() {
    span := rw.span()    // 经过了多少个 interval
    if span <= 0 {
        return
    }
    offset := rw.offset
    for i := 0; i < span; i++ {
        rw.win.resetBucket((offset + i + 1) % rw.size)  // 清零过期桶
    }
    rw.offset = (offset + span) % rw.size
    now := timex.Now()
    rw.lastTime = now - (now-rw.lastTime)%rw.interval   // 对齐到 interval 边界
}

如果距离上次写入经过了 3 个 interval，那就清零接下来的 3 个桶（它们代表的时间段已经过去了），然后将 offset 前移 3 位。lastTime 的对齐操作确保桶的边界是规整的——每个桶始终对齐到 interval 的整数倍时刻。

Reduce 方法遍历桶执行聚合操作：

func (rw *RollingWindow[T, B]) Reduce(fn func(b B)) {
    rw.lock.RLock()
    defer rw.lock.RUnlock()
    var diff int
    span := rw.span()
    if span == 0 && rw.ignoreCurrent {
        diff = rw.size - 1    // 忽略当前桶
    } else {
        diff = rw.size - span
    }
    if diff > 0 {
        offset := (rw.offset + span + 1) % rw.size
        rw.win.reduce(offset, diff, fn)
    }
}

ignoreCurrent 选项的存在是因为"当前桶"中的数据是不完整的——比如一个 1 秒的时间桶刚过了 0.3 秒，此时桶内的数据量只有 30% 的"应该有的量"。如果直接用于计算失败率（失败的请求数 / 总请求数），会因为分母偏小而导致失败率虚高。熔断器使用了 IgnoreCurrentBucket() 选项正是出于这个考虑。

RollingWindow 支持泛型——桶的类型只要实现了 BucketInterface[T] 接口即可：

type BucketInterface[T Numerical] interface {
    Add(v T)
    Reset()
}

默认的 Bucket 实现只记录 Sum 和 Count，分别代表"值总和"和"添加次数"。但如果你需要记录更复杂的数据（比如最大值、最小值、分位数），完全可以实现自己的桶类型。

并发控制：SingleFlight 与 LockedCalls

时间和空间的管理解决了一类问题，但并发编程中还有另一类经典场景：多个 goroutine 同时做同一件事。

SingleFlight：共享调用结果

想象一个高并发场景：缓存过期的那一刻，1000 个请求同时发现缓存未命中，于是 1000 个 goroutine 一起去查数据库——这就是"缓存击穿"。SingleFlight 是它的标准解药：对于同一个 key，只有第一个调用真正执行，后续调用等待并复用第一个的结果。

go-zero 的实现非常紧凑，在 core/syncx/singleflight.go：

type flightGroup struct {
    calls map[string]*call
    lock  sync.Mutex
}

type call struct {
    wg  sync.WaitGroup
    val any
    err error
}

func (g *flightGroup) Do(key string, fn func() (any, error)) (any, error) {
    c, done := g.createCall(key)
    if done {
        return c.val, c.err   // 复用已有结果
    }
    g.makeCall(c, key, fn)    // 第一个调用者执行 fn
    return c.val, c.err
}

createCall 是关键——加锁后在 map 中查找 key，如果已存在则等待 wg.Wait() 返回（等待第一个调用完成）；如果不存在则创建一个新的 call 并 wg.Add(1)：

func (g *flightGroup) createCall(key string) (c *call, done bool) {
    g.lock.Lock()
    if c, ok := g.calls[key]; ok {
        g.lock.Unlock()
        c.wg.Wait()
        return c, true       // done = true，直接返回结果
    }
    c = new(call)
    c.wg.Add(1)
    g.calls[key] = c
    g.lock.Unlock()
    return c, false          // done = false，由当前 goroutine 执行
}

makeCall 在执行完成后清理 map 中的条目并 wg.Done()，唤醒所有等待者：

func (g *flightGroup) makeCall(c *call, key string, fn func() (any, error)) {
    defer func() {
        g.lock.Lock()
        delete(g.calls, key)
        g.lock.Unlock()
        c.wg.Done()
    }()
    c.val, c.err = fn()
}

DoEx 是 Do 的变体，额外返回一个 fresh bool——true 表示这个结果是当前 goroutine 真正执行获得的，false 表示复用了别人的结果。这对 Cache.Take 区分"自己回源"还是"共享结果"非常关键。

LockedCalls：串行化同 key 调用

SingleFlight 是"同 key 调用共享结果"，但有一种不同的需求：同 key 调用必须串行执行，但每次都要独立执行，不共享结果。

一个实际的例子是当你需要保证对同一资源的写操作按顺序执行——并发写入可能导致数据竞争，但你又不能直接复用前面调用的结果，因为每次写入的数据不同。

go-zero 的 LockedCalls 就是为这个场景设计的：

func (lg *lockedGroup) Do(key string, fn func() (any, error)) (any, error) {
begin:
    lg.mu.Lock()
    if wg, ok := lg.m[key]; ok {
        lg.mu.Unlock()
        wg.Wait()          // 等待前一个调用完成
        goto begin         // 然后重新竞争
    }
    return lg.makeCall(key, fn)
}

它的核心是一个 goto begin 循环：如果发现同 key 已有调用在进行，当前 goroutine 等待它完成，然后回到开头重新竞争。这意味着 B 在等待 A 完成后才执行，C 在等待 B 完成后才执行——同 key 调用被串行化了，但每次调用独立执行。

SingleFlight 和 LockedCalls 的区别可以这样理解：

维度	SingleFlight	LockedCalls
执行次数	多个同 key 调用只执行一次	每个同 key 调用都执行
结果共享	后续调用复用第一个的结果	每次调用独立获取结果
用途	防击穿、去重	串行化、防竞态
等待机制	WaitGroup 等待完成	WaitGroup + goto 循环

ResourceManager：SingleFlight 的实际应用

ResourceManager 是一个以 SingleFlight 为核心的资源管理器：

func (manager *ResourceManager) GetResource(key string, create func() (io.Closer, error)) (io.Closer, error) {
    val, err := manager.singleFlight.Do(key, func() (any, error) {
        manager.lock.RLock()
        resource, ok := manager.resources[key]
        manager.lock.RUnlock()
        if ok {
            return resource, nil
        }
        resource, err := create()
        if err != nil {
            return nil, err
        }
        manager.lock.Lock()
        manager.resources[key] = resource
        manager.lock.Unlock()
        return resource, nil
    })
    // ...
}

这里的 SingleFlight 确保同一个资源（比如数据库连接、Redis 客户端）只被创建一次，即使多个 goroutine 同时请求。Close 方法遍历所有资源并调用 Close()，使用 errorx.BatchError 聚合所有错误。

并发原语：Lock、Limit、Pool 与条件等待

go-zero 在标准库 sync 包的基础上，提供了一些更细粒度的并发原语。

SpinLock：当临界区足够短

标准库的 sync.Mutex 在锁竞争时会通过系统调用将 goroutine 挂起——这个挂起和恢复的开销可能比临界区的执行时间还长。当临界区只有几条机器指令时（比如更新一个计数器），自旋锁是更优的选择：

type SpinLock struct {
    lock uint32
}

func (sl *SpinLock) Lock() {
    for !sl.TryLock() {
        runtime.Gosched()    // 自旋失败时让出 CPU
    }
}

func (sl *SpinLock) TryLock() bool {
    return atomic.CompareAndSwapUint32(&sl.lock, 0, 1)
}

SpinLock 使用 CAS 操作尝试获取锁，失败时调用 runtime.Gosched() 让出当前线程的时间片——这比不停重试的空循环（busy loop）要好，因为它给其他 goroutine 运行的机会。

需要注意，SpinLock 不是可重入锁，也没有 Unlock 的持有者检查——任何 goroutine 都可以 Unlock 它。这要求使用者在代码组织上保证成对使用。

Limit 与 TimeoutLimit：限制并发数

Limit 是一个基于 buffered channel 的并发限流器：

type Limit struct {
    pool chan lang.PlaceholderType
}

当 channel 满时，Borrow() 阻塞；TryBorrow() 非阻塞返回 false。Return() 从 channel 取出一个 token 释放。

TimeoutLimit 在 Limit 的基础上增加了超时等待——它组合了我们接下来会讲到的 Cond：

func (l TimeoutLimit) Borrow(timeout time.Duration) error {
    if l.TryBorrow() {
        return nil
    }
    var ok bool
    for {
        timeout, ok = l.cond.WaitWithTimeout(timeout)
        if ok && l.TryBorrow() {
            return nil
        }
        if timeout <= 0 {
            return ErrTimeout
        }
    }
}

循环中的 WaitWithTimeout 会等待其他 goroutine 释放 token 时的 Signal()，但每次醒来后重新竞争 TryBorrow()。剩余超时时间持续减少，直到超时或获取成功。这个实现比给每个等待者分配一个独立的 time.Timer 要轻量得多。

Cond：channel 化的条件变量

go-zero 的 Cond 是 sync.Cond 的 channel 替代：

type Cond struct {
    signal chan lang.PlaceholderType
}

func (cond *Cond) WaitWithTimeout(timeout time.Duration) (time.Duration, bool) {
    timer := time.NewTimer(timeout)
    defer timer.Stop()
    begin := timex.Now()
    select {
    case <-cond.signal:
        return timeout - timex.Since(begin), true
    case <-timer.C:
        return 0, false
    }
}

func (cond *Cond) Signal() {
    select {
    case cond.signal <- lang.Placeholder:
    default:
    }
}

Signal 的非阻塞发送（default 分支）意味着如果没有等待者，信号就丢弃了——这符合条件变量的语义。

Barrier：互斥保护的语法糖

Barrier 是对 sync.Mutex 的最简封装：

type Barrier struct {
    lock sync.Mutex
}

func (b *Barrier) Guard(fn func()) {
    Guard(&b.lock, fn)
}

它本身不提供新能力，但它让代码意图更清晰——当你看到 barrier.Guard(fn)，你立即知道这是在保护一段临界区。它在 PeriodicalExecutor 中用于保护 waitGroup 的并发操作。

Pool：多功能的资源池

sync.Pool 适用于临时对象的复用（没有数量上限，随时可能被 GC 回收），但业务场景中往往需要更多控制：数量上限、最大存活时间、自定义销毁逻辑。go-zero 的 Pool 提供了这三个能力：

type Pool struct {
    limit   int              // 最大资源数
    created int              // 当前已创建数
    maxAge  time.Duration    // 资源最大存活时间
    lock    sync.Locker
    cond    *sync.Cond
    head    *node            // 空闲资源链表
    create  func() any
    destroy func(any)
}

Get 方法的核心逻辑：

func (p *Pool) Get() any {
    p.lock.Lock()
    defer p.lock.Unlock()
    for {
        if p.head != nil {
            head := p.head
            p.head = head.next
            if p.maxAge > 0 && head.lastUsed+p.maxAge < timex.Now() {
                p.created--
                p.destroy(head.item)   // 过期资源，销毁后继续循环
                continue
            } else {
                return head.item       // 返回空闲资源
            }
        }
        if p.created < p.limit {
            p.created++
            return p.create()          // 创建新资源
        }
        p.cond.Wait()                  // 等待资源归还
    }
}

当一个资源被取出时检查了 maxAge——空闲时间超过上限的资源会被销毁而不是复用。这解决了连接池中"僵尸连接"的问题——一个 TCP 连接长时间不用可能已经被对端关闭了，复用它会直接失败。destroy 回调给了调用者一个清理钩子（比如关闭底层连接）。

执行器体系：从延时到批量

单次的任务分发很简单——go fn() 就行了。但实际项目中，任务往往带有约束：需要延时执行、需要批量攒够一批再执行、需要按固定的时间周期执行、需要限制执行频率……go-zero 的 core/executors/ 包为这些需求提供了一套递进的抽象。

基础抽象：TaskContainer 与 Execute

所有执行器都围绕两个接口展开：

type Execute func(tasks []any)

type TaskContainer interface {
    AddTask(task any) bool     // 返回 true 表示需要立即 flush
    Execute(tasks any)          // 执行收集到的任务
    RemoveAll() any             // 取出所有任务并清空
}

AddTask 返回的 bool 值是一个关键设计——当容器"满了"时返回 true，通知上层的 PeriodicalExecutor 立即 flush，不用等定时器触发。

LessExecutor：最简单的频率限制器

我们先从最简单的开始。LessExecutor 只做一件事：在指定时间间隔内最多执行一次：

func (le *LessExecutor) DoOrDiscard(execute func()) bool {
    now := timex.Now()
    lastTime := le.lastTime.Load()
    if lastTime == 0 || lastTime+le.threshold < now {
        le.lastTime.Set(now)
        execute()
        return true
    }
    return false
}

我们在第十一篇文章中见过它——日志的 limitedExecutor 就是基于同样的思路，防止同一错误日志在 100ms 内重复输出成千上万次。

DelayExecutor：延迟触发且合并多次触发

LessExecutor 解决的是"限频"，DelayExecutor 解决的是"防抖"。当 Trigger 被多次调用时，只有第一次会启动定时器，后续调用被忽略：

func (de *DelayExecutor) Trigger() {
    de.lock.Lock()
    defer de.lock.Unlock()
    if de.triggered {
        return    // 已触发，忽略
    }
    de.triggered = true
    threading.GoSafe(func() {
        timer := time.NewTimer(de.delay)
        defer timer.Stop()
        <-timer.C
        de.lock.Lock()
        de.triggered = false
        de.lock.Unlock()
        de.fn()
    })
}

注意 triggered 标志在 fn() 之前就复位了——这确保如果在 fn 执行期间又有新的 Trigger，新的定时器能正常启动，不会因为 triggered = true 而被忽略。

PeriodicalExecutor：执行器体系的中枢

PeriodicalExecutor 是整个执行器体系的心脏，BulkExecutor 和 ChunkExecutor 都是它的包装：

type PeriodicalExecutor struct {
    commander   chan any
    interval    time.Duration
    container   TaskContainer
    waitGroup   sync.WaitGroup
    wgBarrier   syncx.Barrier
    confirmChan chan lang.PlaceholderType
    inflight    int32
    guarded     bool
    newTicker   func(duration time.Duration) timex.Ticker
    lock        sync.Mutex
}

它的核心工作流是这样的：

1. Add(task) 调用 container.AddTask(task)。如果容器返回 true（满了），则通过 commander channel 通知后台 goroutine 立即 flush，然后等待 confirmChan 确认——这个同步等待确保任务被处理后 Add 才返回。
2. 如果不是第一次 Add，且后台 goroutine 还没启动，则启动 backgroundFlush。
3. backgroundFlush 是一个事件循环，监听 commander（立即 flush 的命令）和 ticker（定期 flush）：

func (pe *PeriodicalExecutor) backgroundFlush() {
    go func() {
        defer pe.Flush()
        ticker := pe.newTicker(pe.interval)
        defer ticker.Stop()
        var commanded bool
        last := timex.Now()
        for {
            select {
            case vals := <-pe.commander:
                commanded = true
                atomic.AddInt32(&pe.inflight, -1)
                pe.enterExecution()
                pe.confirmChan <- lang.Placeholder
                pe.executeTasks(vals)
                last = timex.Now()
            case <-ticker.Chan():
                if commanded {
                    commanded = false
                } else if pe.Flush() {
                    last = timex.Now()
                } else if pe.shallQuit(last) {
                    return   // 空闲退出
                }
            }
        }
    }()
}

几个精妙的设计细节：

Idle 退出。 shallQuit 检查上次活动距今是否超过 interval * 10，且没有 inflight 等待的任务。如果都满足，后台 goroutine 退出（guarded = false），下次 Add 时再重新启动。这意味着在低流量时段，不会有一个"空转"的 goroutine 白白消耗资源。

inflight 计数。 当积压的任务被取出等待处理时（还未经 executeTasks 执行），inflight 计数持续为 1。这确保即使定时器触发时发现容器为空，只要还有 inflight 的任务，后台 goroutine 就不会退出——在任务从"取出"到"被执行"之间有一个窗口期。

优雅退出时的 flush。 构造函数中注册了 proc.AddShutdownListener(func() { executor.Flush() })，确保进程退出前最后一批任务不丢失。backgroundFlush 的 defer pe.Flush() 也是同理——goroutine 退出时做最后一次清理。

BulkExecutor 与 ChunkExecutor：两种聚合策略

BulkExecutor 和 ChunkExecutor 都是 PeriodicalExecutor 的薄包装，区别在于"满"的判定标准：

• BulkExecutor：按任务数量判定。容器内的任务数达到 cachedTasks（默认 1000）时触发 flush。适用于日志批量写入、事件批量发送等场景。
• ChunkExecutor：按数据大小判定。每个 Add(task, size) 传入该任务的数据大小，容器内的累计大小达到 maxChunkSize（默认 1MB）时触发 flush。适用于批量传输需要控制单次数据量的场景。

两者的执行都复用 PeriodicalExecutor 的定时 flush 机制——即使不到阈值，每隔 flushInterval 也会执行一次。

高级抽象：Stream、MapReduce 与并行控制

有了执行器这套"积木"，go-zero 在更上层还提供了两套面向不同场景的抽象：fx 包（函数式工具）和 mr 包（MapReduce 模式）。

fx.Stream：惰性求值的管道式处理

fx.Stream 受 ReactiveX 启发，提供了一套管道式处理 API：

fx.Just(1, 2, 3, 4, 5).
    Filter(func(item any) bool { return item.(int)%2 == 0 }).
    Map(func(item any) any { return item.(int) * 10 }).
    ForEach(func(item any) { fmt.Println(item) })

Stream 的所有操作都是惰性的——它们不立即执行，而是构建一个处理管道，只有终端操作（ForEach、ForAll、Reduce、Done 等）才会真正拉动数据。

Walk 是 Stream 的核心方法，其他转换操作（Filter、Map）都是通过它实现的：

func (s Stream) Walk(fn WalkFunc, opts ...Option) Stream {
    option := buildOptions(opts...)
    if option.unlimitedWorkers {
        return s.walkUnlimited(fn, option)
    }
    return s.walkLimited(fn, option)
}

walkLimited 使用 buffered channel 作为 worker 池，最多 workers（默认 16）个 goroutine 并发处理：

func (s Stream) walkLimited(fn WalkFunc, option *rxOptions) Stream {
    pipe := make(chan any, option.workers)
    go func() {
        var wg sync.WaitGroup
        pool := make(chan lang.PlaceholderType, option.workers)
        for item := range s.source {
            val := item
            pool <- lang.Placeholder
            wg.Add(1)
            threading.GoSafe(func() {
                defer func() { wg.Done(); <-pool }()
                fn(val, pipe)
            })
        }
        wg.Wait()
        close(pipe)
    }()
    return Range(pipe)
}

一个容易忽略的细节是顺序性——walkLimited 不保证输出的顺序与输入一致。因为多个 worker 并发处理，先完成的 worker 会先把结果写入 pipe。walkUnlimited 同理。如果你需要保持顺序，应该用串行的 Walk。

Head 方法的实现也值得一看——取前 N 个元素后，它不会直接 break 循环（那样会让上游 goroutine 永久阻塞），而是启动一个后台 goroutine drain(s.source) 把剩余元素全部消费掉：

if n == 0 {
    close(source)
    drain(s.source)   // 防止上游 goroutine 泄漏
}

这个模式在 First、AllMatch、AnyMatch 等方法中反复出现——凡是提前终止的操作，都需要 drain 上游 channel 防止 goroutine 泄漏。

fx 并行与超时工具

fx.Parallel 和 fx.ParallelErr 是对 RoutineGroup 的最简封装——并行执行多个函数并等待全部完成。ParallelErr 额外收集所有错误到 BatchError。

DoWithTimeout 为任意函数增加超时控制：

func DoWithTimeout(fn func() error, timeout time.Duration, opts ...DoOption) error {
    parentCtx := context.Background()
    for _, opt := range opts {
        parentCtx = opt()
    }
    ctx, cancel := context.WithTimeout(parentCtx, timeout)
    defer cancel()
    done := make(chan error, 1)
    panicChan := make(chan any, 1)
    go func() {
        defer func() {
            if p := recover(); p != nil {
                panicChan <- fmt.Sprintf("%+v\n\n%s", p, strings.TrimSpace(string(debug.Stack())))
            }
        }()
        done <- fn()
    }()
    select {
    case p := <-panicChan:
        panic(p)
    case err := <-done:
        return err
    case <-ctx.Done():
        return ctx.Err()
    }
}

注意 done channel 的缓冲区大小为 1——如果超时先触发，ctx.Done() 返回，函数不再等待 done。但 fn 所在的 goroutine 还在运行（无法被强制终止），当它最终完成时会向 done 写入——如果没有缓冲区，这个写入会永久阻塞，导致 goroutine 泄漏。

DoWithRetry 提供了带间隔的重试逻辑，支持通过 WithRetry 设置重试次数（默认 3 次）、通过 WithInterval 设置重试间隔、通过 WithTimeout 设置总超时。一个特别的设计是 WithIgnoreErrors——如果错误匹配指定类型，直接返回 nil 而不重试：

for _, ignoreErr := range options.ignoreErrors {
    if errors.Is(err, ignoreErr) {
        return nil
    }
}

MapReduce：带取消与错误传播的并行处理

core/mr/ 是 go-zero 中最精巧的并发抽象之一。它与 fx.Stream 不同——Stream 是管道式、链式调用的，MapReduce 是一次性配置好整个处理流程然后执行。

MapReduce 函数的签名浓缩了框架的全部能力：

func MapReduce[T, U, V any](
    generate GenerateFunc[T],
    mapper  MapperFunc[T, U],
    reducer ReducerFunc[U, V],
    opts ...Option,
) (V, error)

三个阶段的职责分别是：

• Generate：向 source channel 发送输入元素
• Mapper：并发处理每个元素，将结果写入 collector channel
• Reducer：从 collector channel 读取所有 mapper 输出，聚合为一个最终结果

比标准 MapReduce 多出来的是取消传播。MapperFunc 和 ReducerFunc 都接收一个 cancel func(error)——mapper 或 reducer 可以在任何时候取消整个处理流程：

type MapperFunc[T, U any] func(item T, writer Writer[U], cancel func(error))
type ReducerFunc[U, V any] func(pipe <-chan U, writer Writer[V], cancel func(error))

取消的实现通过 GuardedWriter + done channel：

type guardedWriter[T any] struct {
    ctx     context.Context
    channel chan<- T
    done    <-chan struct{}
}

func (gw guardedWriter[T]) Write(v T) {
    select {
    case <-gw.ctx.Done():
    case <-gw.done:
    default:
        gw.channel <- v
    }
}

一旦 cancel 被调用，done channel 关闭，所有正在执行的 mapper 在下次 Write 时发现 done 已关闭，从而停止写入——优雅地结束而非粗暴终止。同时 executeMappers 中的主循环也会因为 doneChan 关闭而退出：

for atomic.LoadInt32(&failed) == 0 {
    select {
    case <-mCtx.ctx.Done():
        return
    case <-mCtx.doneChan:
        return
    case pool <- struct{}{}:
        // ...
    }
}

mapperContext 中还有一个 panicChan——当任何一个 mapper panic 时，panic 信息通过 onceChan 只发送一次（CAS 保护），然后在主 goroutine 中重新 panic：

defer func() {
    if r := recover(); r != nil {
        atomic.AddInt32(&failed, 1)
        mCtx.panicChan.write(buildPanicInfo(r, debug.Stack()))
    }
}()

这种设计使得 MapReduce 中的 panic 不会被静默吞噬——它在主 goroutine 中重现，保留完整的调用栈。

Finish 和 FinishVoid 是两个快捷函数。Finish 并行执行多个 func() error，任一报错即取消其他：

func Finish(fns ...func() error) error {
    return MapReduceVoid(func(source chan<- func() error) {
        for _, fn := range fns { source <- fn }
    }, func(fn func() error, writer Writer[any], cancel func(error)) {
        if err := fn(); err != nil { cancel(err) }
    }, func(pipe <-chan any, cancel func(error)) {
    }, WithWorkers(len(fns)))
}

线程工具：Goroutine 生命周期管理

最后看 core/threading/ 包，它提供了一系列 goroutine 管理的"安全网"。

GoSafe 与 RunSafe：Panic 不扩散

func RunSafe(fn func()) {
    defer rescue.Recover()
    fn()
}

生产代码中，任何启动新 goroutine 的地方都应该使用 GoSafe 或 RunSafe——如果 goroutine 中发生了未捕获的 panic，它会将整个进程崩溃。rescue.Recover() 捕获 panic 并记录堆栈，使问题可排查但不影响其他 goroutine。

RoutineGroup：WaitGroup 的并发安全包装

RoutineGroup 是对 sync.WaitGroup 的"防误用"包装——Add 和 go func() { defer Done() } 被封装在 Run 和 RunSafe 中，调用方不会忘记成对使用：

func (g *RoutineGroup) Run(fn func()) {
    g.waitGroup.Add(1)
    go func() {
        defer g.waitGroup.Done()
        fn()
    }()
}

TaskRunner：有并发上限的 Goroutine 调度器

TaskRunner 通过 buffered channel 限制同时运行的 goroutine 数量：

func (rp *TaskRunner) Schedule(task func()) {
    rp.waitGroup.Add(1)
    rp.limitChan <- lang.Placeholder   // 满时阻塞
    go func() {
        defer rescue.Recover(func() {
            <-rp.limitChan
            rp.waitGroup.Done()
        })
        task()
    }()
}

注意 waitGroup.Add(1) 在 limitChan <- lang.Placeholder 之前——这是有意为之。如果顺序反过来，可能出现这种情况：limitChan 满了，任务在排队，但 Wait() 被调用了，因为 waitGroup 计数为 0，Wait() 立即返回——而实际上任务还在等待调度。

StableRunner：保序的并发处理

go-zero 中最有趣的并发结构是 StableRunner——它解决了"并发处理但顺序输出"的问题。这类似于 Kafka 消费者的常见需求：消息可以并发处理以提升吞吐，但提交 offset 时必须按原始顺序：

type StableRunner[I, O any] struct {
    handle        func(I) O
    consumedIndex uint64
    writtenIndex  uint64
    ring          []*struct {
        value chan O
        lock  sync.Mutex
    }
    runner *TaskRunner
    done   chan struct{}
}

它的工作原理是：

1. Push(v) 为消息分配一个递增的序号（writtenIndex），放进环形数组对应位置（index % bufSize）
2. 通过 TaskRunner 调度并发处理，结果写入对应位置的 value channel
3. Get() 按序号顺序（consumedIndex）读取 value channel——即使后面的消息先处理完，Get() 也会阻塞等待前面的消息

每个位置有一个 lock 确保同一位置不会被并发 Push，而 Get() 的调用者必然是单 goroutine（文档上明确要求）。

总结

本文从"上层建筑下面藏着什么"开始，梳理了 go-zero 的时间与并发工具箱。全文没有给出各组件之间一张复杂的依赖图——因为这些工具的关系并不是树形的依赖层级，而是分层的组合关系：

• 最底层是原子类型和 channel：AtomicDuration、AtomicBool、SpinLock，以及基于 buffered channel 的 Limit 和 Cond
• 组合层是用底层工具构建的复合结构：TimeoutLimit = Limit + Cond；Pool = sync.Mutex + sync.Cond + 链表
• 时间层是专门处理时间维度的组件：TimingWheel（未来任务调度）、RollingWindow（过去数据统计）
• 协调层是管理 goroutine 间协作的机制：SingleFlight（共享结果）、LockedCalls（串行化）、Barrier（互斥保护）
• 执行器层是任务调度策略：PeriodicalExecutor → BulkExecutor/ChunkExecutor；LessExecutor；DelayExecutor
• 抽象层是面向特定模式的高级工具：fx.Stream（管道处理）、MapReduce（并行聚合）、StableRunner（保序并发）