# consumer 与背压 · MutatesData 的精妙

> 组件间怎么传数据？一个只有两个方法的接口。这一章读 consumer 与 fanoutconsumer 源码，看那个 MutatesData 契约怎么让「扇出时只给要改数据的人复制」，以及背压——Collector 最难也最重要的工程问题。

- 作者：David（道雾轩）
- 专栏：指标采集与观测（https://daiw.org/manual/observability.md）
- 最后更新：2026-07-15
- 原文：https://daiw.org/manual/observability/consumer-backpressure
- 转载与引用：请注明出处并附原文链接（https://daiw.org/about/copyright）

# consumer 与背压 · MutatesData 的精妙

[组件不知道下游是谁](https://daiw.org/manual/observability/collector-architecture)，它只拿到一个 **`consumer`** 接口往里塞数据。这一章看这个接口，以及它背后两个真正的工程难题：**避免不必要的复制**和**背压**。

## 接口小到极致

```go
// consumer/metrics.go:15
type Metrics interface {
    internal.BaseConsumer
    // ConsumeMetrics processes the metrics. After the function returns, the metrics
    // are no longer accessible, and accessing them is considered undefined behavior.
    ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error
}

// consumer/internal/consumer.go:16
type BaseConsumer interface {
    Capabilities() Capabilities
}
```

**两个方法：`ConsumeMetrics`（收数据）和 `Capabilities`（声明我会不会改数据）。** [traces/logs 各有对应的接口](https://daiw.org/manual/observability/pdata)，形状一样。

<Callout type="info">
  **注意 `ConsumeMetrics` 的注释：“After the function returns, the metrics are **no longer accessible**, and accessing them is considered **undefined behavior**.”**

  **这是个很硬的契约**：函数返回后，你**不能再碰**这批数据。为什么这么严格？因为它给了上游**回收 / 复用内存**的自由——如果 consumer 可以偷偷留一份引用晚点再用，上游就永远不敢复用缓冲区。

  **想异步处理？你必须自己复制一份。** 这个约束把“谁负责复制”说得清清楚楚，消除了一整类“[数据被上游改了 / 被回收了](https://daiw.org/manual/observability/pdata)”的诡异 bug。
</Callout>

## MutatesData：一个字段的大用

```go
// consumer/internal/consumer.go:7(真实注释)
// Capabilities describes the capabilities of a Processor.
type Capabilities struct {
    // MutatesData is set to true if Consume* function of the
    // processor modifies the input Traces, Logs or Metrics argument.
    // Processors which modify the input data MUST set this flag to true. If the processor
    // does not modify the data it MUST set this flag to false. If the processor creates
    // a copy of the data before modifying then this flag can be safely set to false.
    MutatesData bool
}
```

<Callout type="info">
  **每个组件必须**诚实申报**：“我会不会改这批数据？” 这个 bool 看着微不足道，却是[整个扇出优化](https://daiw.org/manual/observability/collector-architecture)的基础。**

  回忆[上一章](https://daiw.org/manual/observability/pdata)：pdata **禁止可变共享**，所以扇出时“安全”的做法是**给每条下游管道都复制一份**。但**复制是有成本的**（高吞吐下非常可观）。

  **`MutatesData` 让 Collector 能做到“只在必要时复制”。** 看 `internal/fanoutconsumer/metrics.go` 的真实注释：

  > NewMetrics wraps multiple metrics consumers in a single one. It fans out the incoming data to all the consumers, and does **smart routing**:
  > - **Clones only to the consumer that needs to mutate the data.**
  > - **If all consumers needs to mutate the data one will get the original mutable data.**
</Callout>

## 扇出的智能路由

看实现（`fanoutconsumer/metrics.go`，真实代码）：

```go
// 只有一个消费者且它不改数据 → 直接返回它,连包装都不要!
if len(mcs) == 1 && !mcs[0].Capabilities().MutatesData {
    return mcs[0]
}
// 否则:把消费者分成两拨
for i := range mcs {
    if mcs[i].Capabilities().MutatesData {
        mc.mutable = append(mc.mutable, mcs[i])    // 会改的
    } else {
        mc.readonly = append(mc.readonly, mcs[i])  // 只读的
    }
}
```

<Callout type="info">
  **分类之后，复制策略就清楚了：**

  ```
  假设扇出给 4 个消费者:2 个会改(mutable)、2 个只读(readonly)

  ① 给第 1 个 mutable   → 复制一份(它要改,不能碰原件)
  ② 给第 2 个 mutable   → 原件直接给它(反正后面没人要了)★
  ③ 给 2 个 readonly    → 共享同一份,标记为 MarkReadOnly() ★
  ```

  **两处精妙：**

  - **“最后一个 mutable 消费者拿原件”** —— 注释说 “Clone the data before sending to all mutating consumers **except the last one**”。前面的都要复制，但**最后一个不用**——因为它改了也没人再看了。**省下一次复制。**
  - **`MarkReadOnly()`** —— 多个只读消费者**共享同一份**数据，但把它**标记成只读**。于是它们**安全地共享**，零复制。这本质是[写时复制思想的变体](https://daiw.org/manual/webrtc/buffers)——**用“声明 + 标记”代替“每人一份”**。

  还有个细节：`Capabilities()` 自己也会**向上传播**：
  ```go
  return consumer.Capabilities{MutatesData: len(msc.mutable) > 0 && len(msc.readonly) == 0}
  ```
  **一个扇出节点“会不会改数据”，取决于它下游的组成**——这样整个图的复制决策能**逐层推导**出来。

  **一个 bool 字段，换来整条管道的零冗余复制。** 这是“**让组件申报意图，框架据此优化**”的漂亮范例——[和 `storage.Querier` 的 `sortSeries`/`hints` 是一个路子](https://daiw.org/manual/observability/query-path)：**框架不猜，让调用方说**。
</Callout>

<Callout type="warn">
  **代价是：这个契约靠**自觉**。** 注释里那个 **MUST** 是大写的——“Processors which modify the input data **MUST** set this flag to true”。

  **如果你写了个 processor，改了数据却申报 `MutatesData: false`**——编译器不会拦你，测试可能也发现不了，但**线上会出现“另一条管道的数据莫名其妙被改了”**这种极难查的 bug。

  **[写自定义组件时](https://daiw.org/manual/observability/custom-components)，这是最该小心的一条。** 拿不准就填 `true`（牺牲一点性能，换正确性）。
</Callout>

## 背压：最难的那个问题

<Callout type="warn">
  **`ConsumeMetrics` 返回 `error`，而且它是**同步**的。这个设计决定了 Collector 的背压模型：**

  ```
  receiver → processor → exporter → 后端(慢了!)
                                      ↓
                         exporter 的 Consume 阻塞/返回错误
                                      ↓
                         processor 的 Consume 也阻塞
                                      ↓
                         receiver 的 Consume 也阻塞
                                      ↓
                         receiver 对客户端返回错误/拒绝
  ```

  **背压天然沿着调用栈往回传**——这是同步接口的直接后果，也是它的优点：**压力被如实传导到源头**，而不是在中间某处悄悄丢数据。

  **但这带来一个残酷的取舍：后端慢了，你要怎么办？**

  | 策略 | 后果 |
  | --- | --- |
  | **阻塞** | 压力传回应用，**可能拖慢业务** |
  | **缓冲** | 内存涨，**缓冲满了还是要选** |
  | **丢数据** | 保住了系统，**但监控有洞** |

  **没有免费的选择。** Collector 的做法是给你**可配置的组件**：
  - **`batch` processor** —— 攒批发送，**削峰**（但也增加延迟）。
  - **`memory_limiter` processor** —— **内存超限时主动拒绝新数据**（返回错误 → 背压传回去）。**这是保命的**：没有它，后端一慢，Collector 就会无限缓冲直到 OOM——**然后所有数据都丢了**。
  - **exporter 的 `sending_queue` + `retry_on_failure`** —— 队列 + 重试，**队列满了就丢**。

  **`memory_limiter` 和 [Prometheus 的 `sample_limit`](https://daiw.org/manual/observability/scrape-loop)、[`MaxSamples`](https://daiw.org/manual/observability/promql-engine) 是同一个哲学**：**宁可明确地拒绝，也不要含糊地崩掉**。**这三个东西出现在三个完全不同的地方，却是同一条工程原则——值得记住。**

  **实战建议：`memory_limiter` 应该放在管道的**最前面**（紧跟 receiver）。** 因为它要在数据**进入**管道前就拦住，而不是等它们已经在 processor 里堆了一堆了才发现内存爆了。
</Callout>

## 小结与下一章

组件间靠 **`consumer` 接口**传数据，小到只有两个方法：`ConsumeMetrics` + `Capabilities`。**`ConsumeMetrics` 的契约很硬**：[返回后不得再碰数据（undefined behavior）——这给了上游复用内存的自由，想异步就自己复制](https://daiw.org/manual/observability/pdata)。**`MutatesData` 是精妙所在**：[每个组件诚实申报“我改不改数据”，框架据此**智能路由**——只给要改的复制、**最后一个 mutable 拿原件**（省一次复制）、多个只读**共享 + `MarkReadOnly()`**（零复制）；`Capabilities` 还会逐层向上传播，让整图的复制决策可推导](https://daiw.org/manual/webrtc/buffers)。**一个 bool 换来整条管道零冗余复制**——[和 `Querier` 的 `sortSeries`/`hints` 一个路子：框架不猜，让调用方说](https://daiw.org/manual/observability/query-path)；**代价是靠自觉**（[申报错了会出极难查的 bug，拿不准填 `true`](https://daiw.org/manual/observability/custom-components)）。**背压**因接口同步而[天然沿调用栈回传到源头](https://daiw.org/manual/observability/consumer-backpressure)——后端慢时[阻塞/缓冲/丢数据三选一，没有免费选择](https://daiw.org/manual/observability/consumer-backpressure)；**`memory_limiter` 是保命的**（[超限主动拒绝，否则无限缓冲直到 OOM 然后全丢](https://daiw.org/manual/observability/scrape-loop)），该放在管道**最前面**。它和 [`sample_limit`](https://daiw.org/manual/observability/scrape-loop)、[`MaxSamples`](https://daiw.org/manual/observability/promql-engine) 是**同一条原则：宁可明确拒绝，不要含糊崩掉**。

组件都认识了，怎么把它们配起来？下一章看配置与部署形态。👉 [配置与部署](https://daiw.org/manual/observability/collector-config)。

## 参考资料

- [1] OTel Collector 源码 `consumer/metrics.go:15`、`consumer/internal/consumer.go:7`（`Capabilities`/`MutatesData` 注释）。
- [2] OTel Collector 源码 `internal/fanoutconsumer/metrics.go`（smart routing、`MarkReadOnly`、最后一个 mutable 拿原件）。
- [3] 关联[pdata 避免可变共享](https://daiw.org/manual/observability/pdata)、[sample_limit 的同一哲学](https://daiw.org/manual/observability/scrape-loop)。