异步流水线工程：BatchProcessor

本文是《从 0 写 Agent》系列第 26 章。这个系列基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版），用 Go 从零实现一个生产级 Agent 运行时；本章讲「BatchProcessor：异步流水线工程」。

Tracing 不能拖慢 Agent 循环，你不能因为"记个 trace"让 Agent 变慢。所以 trace 要异步处理。

agents-go 的 BatchProcessor 是个异步流水线，后台 goroutine 缓冲 trace/span，批量导出。它处理了背压、孤儿 span 防护、优雅关闭三个工程难题。这一章拆这个"小型消息队列"。

引子：为什么 trace 要异步

朴素做法，每产生一个 span，立即调 exporter 导出（发 HTTP、写文件）。问题：

• 阻塞 Agent 循环，网络导出可能几百毫秒，Agent 循环要等
• 频繁 IO，100 个 span 发 100 次 HTTP，效率低
• 失败影响业务，trace 系统挂了，Agent 也挂了

正确做法，trace 异步。 Agent 循环把 span 丢给一个队列就返回（微秒级），后台 goroutine 慢慢批量导出。trace 系统的问题不波及 Agent。

源码对照：BatchProcessor 的架构

BatchProcessor（tracing/batch_processor.go）是个经典的生产者-消费者 + 批处理：

Agent 循环（生产者）           后台 goroutine（消费者）
    │                              │
    ├─ OnSpanStart(span)           │
    ├─ OnSpanEnd(span)   ─► queue  ├─ 从 queue 取
    │                      (channel) ├─ 攒到 MaxBatchSize 或超时
    │                              ├─ 调 exporter.Export(batch)
    │                              └─ 循环

三个触发导出的条件

1. 定时器，每 FlushInterval 导出一次（默认几秒）
2. 攒够了，队列里达到 MaxBatchSize，立即导出
3. 关闭，Shutdown 时导出剩余的

flushNow 信号，非阻塞通知

当队列达到 MaxBatchSize，要通知后台"该导出了"。用 buffered channel（batch_processor.go:74）：

// 概念示意
select {
case bp.flushNow <- struct{}{}:  // 发通知（buffer=1，不阻塞）
default:                          // 已经有通知了，不重复发
}

buffer=1 + default 分支，如果已经有一个待处理的 flush 通知，不发第二个。这防止"快速产生 span 时，flush channel 被塞满"。

防孤儿 span：trace start 即入队

一个精妙的设计，trace 在 start 时就入队，不是 end 时（batch_processor.go:83）。

为什么？考虑这个场景：

1. Agent 循环开始一个 trace
2. Agent 循环跑到一半崩溃了（panic、OOM）
3. trace 没 end，span 没 end

如果 trace 只在 end 入队，崩溃的 trace 永远不会被导出，成了"孤儿"，你根本不知道它存在过。

trace start 即入队，保证即使中途崩溃，至少 trace 的"开始"被记录了。 你能看到"这个 trace 开始了但没结束"，知道有过这么一次失败的 Run。

注释明说（batch_processor.go:83），这是对标 Python SDK 的设计，防 run 中途崩溃导致 span 孤儿。

背压：超 MaxQueueSize 丢弃并计数

如果导出很慢，队列会涨。不能无限涨（内存爆）。BatchProcessor 的策略，超 MaxQueueSize 时丢弃，并计数（batch_processor.go:65）：

// 概念示意
if len(queue) >= maxQueueSize {
    atomic.AddInt64(&dropped, 1)  // 丢弃计数
    return  // 不入队
}
queue <- span

丢弃而非阻塞，因为 trace 是"best-effort"的可观测数据，不应该因为队列满而阻塞 Agent 循环。丢弃计数让你知道"丢了多少"，方便告警。

这是"telemetry 不应该破坏应用"原则的体现，宁可丢 trace，也不能让 Agent 卡住。

优雅关闭

Shutdown（batch_processor.go:137）要保证，关闭时，队列里剩余的 trace 都被导出。但不能无限等：

// 概念示意
func (bp *BatchProcessor) Shutdown(ctx context.Context) {
    bp.stopOnce.Do(func() {
        close(bp.done)  // 通知后台停止
    })
    bp.wg.Wait()  // 等后台 goroutine 退出

    // 配合 ctx：超时就放弃等待
    select {
    case <-bp.done:
    case <-ctx.Done():
    }
}

stopOnce，保证 close(done) 只调一次（多次 close channel 会 panic）。wg.Wait()，等后台 goroutine 把剩余的导出完。ctx 超时，如果导出太慢，放弃等待（不让 Shutdown 永远阻塞）。

这是异步系统关闭的标准范式，通知停止 + 等待完成 + 超时兜底。

设计权衡：BatchProcessor 的参数

调参原则，如果 trace 是"实时监控"用，小 batch + 短 interval；如果是"事后分析"用，大 batch + 长 interval。根据你的 exporter 速度调。

小结

读完这一章，你应该理解 BatchProcessor 的三个工程要点：

1. 异步批处理，后台 goroutine，Agent 循环不阻塞
2. trace start 即入队，防崩溃导致孤儿 span
3. 背压丢弃 + 优雅关闭，telemetry 不破坏应用，但要有计数和超时

BatchProcessor 是个微型消息队列，生产者（Agent 循环）、队列（channel）、消费者（后台 goroutine）、背压（丢弃）、关闭（graceful shutdown）。 理解它，你就理解了异步系统的核心模式。

下一篇

第 27 章，Exporter：把 trace 送出去。Noop/Func/Console/Collecting/HTTP 五种 exporter，各有用途。其中 HTTPExporter 有个重要原则，"失败静默丢弃不重试"，telemetry 不能反过来搞挂应用。

本文是《从 0 写 Agent》系列第 26 章，案例代码基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版）。全系列目录见总览。