异步化与背压控制

异步化是分布式系统提升吞吐与解耦的核心手段，但异步如果不加约束，就会产生"生产快、消费慢"的背压（Backpressure）问题：队列无限积压、内存被打爆、下游被冲垮。本文从异步通信模式讲到背压原理，再到流量整形与生产实践，给出异步系统的完整稳定性设计。

1. 异步通信模式

1.1 同步 vs 异步

1.2 常见异步模式

1. 消息队列：发送方投递消息，消费者异步处理（Kafka/RabbitMQ/RocketMQ）
2. 事件驱动：业务事件发布/订阅，消费者响应（见 https://plumephp.com/distributed-event-driven-architecture/）
3. 回调/CompletableFuture：发起异步调用，完成时回调
4. 响应式流：Publisher/Subscriber 模型，带背压协议（Reactor/RxJava）
5. 本地异步（线程池/队列）：进程内任务解耦

// CompletableFuture 异步编排
CompletableFuture<OrderResult> future = CompletableFuture
    .supplyAsync(() -> orderService.create(request))          // 异步执行
    .thenApply(order -> paymentService.charge(order))         // 链式
    .exceptionally(e -> { log.error("create order failed", e);
                          return OrderResult.failed(e); });

1.3 异步系统的核心矛盾

异步引入后，生产速率 ≠ 消费速率。当生产者持续快于消费者，队列/内存中积压的数据不断增长，最终触发 OOM 或下游过载。背压就是解决这一矛盾的机制。

2. 背压原理

2.1 什么是背压

背压（Backpressure）：当下游处理能力不足时，上游感知到压力并放慢生产速度的机制。物理世界的水管最有代表性：下游堵住，水流就反向顶回去。

无背压：生产者 1000 msg/s  ──► 无界队列（OOM 风险） ◄── 消费者 100 msg/s
有背压：生产者 1000 msg/s  ──► 有界队列 + 满时阻塞/丢弃 ◄── 消费者 100 msg/s
                               消费者慢了，生产者被"顶住"

2.2 TCP 背压：滑动窗口

TCP 是最经典的背压实现。发送方按接收窗口（rwnd）控制未确认数据量，接收方缓冲区满时缩小窗口，发送方自然放慢：

发送方 ──► 已发送未确认数据 ≤ 接收窗口 rwnd
接收方缓冲区紧张 → 通告更小窗口 → 发送方放慢

2.3 消息队列的背压

MQ 天然缓冲解耦，但背压体现在消费积压：

Kafka 消费者通过 max.poll.records 与 max.poll.interval.ms 控制每轮拉取量，本质上就是拉模式的背压。

2.4 响应式流的背压

Reactive Streams 规范定义了基于**请求量（Demand）**的背压：订阅者告诉发布者"我现在还能处理 n 个"，发布者最多发 n 个。

Publisher ──request(10)──► Subscriber
Publisher ──onNext ×10──► Subscriber   ← 只发请求的量
Subscriber 处理完 → 再 request(n)

// Reactor 背压：下游只请求上游一次发射 5 个元素
Flux.range(1, 1_000_000)
    .limitRate(5)                       // 每次请求 5 个，动态补单
    .doOnNext(i -> slowProcess(i))      // 下游处理很慢
    .subscribe();

3. 流量整形与削峰

3.1 令牌桶（Token Bucket）

以恒定速率向桶中放令牌，请求需要消耗令牌才放行。允许突发（桶容量 = 突发量），长期速率受控。

type TokenBucket struct {
    rate     float64  // 每秒放令牌数
    capacity float64  // 桶容量（最大突发）
    tokens   float64
    mu       sync.Mutex
    lastRefill time.Time
}

func (b *TokenBucket) Allow() bool {
    b.mu.Lock()
    defer b.mu.Unlock()
    now := time.Now()
    // 按时间补充令牌
    b.tokens = math.Min(b.capacity,
        b.tokens + now.Sub(b.lastRefill).Seconds()*b.rate)
    b.lastRefill = now
    if b.tokens < 1 {
        return false
    }
    b.tokens--
    return true
}

3.2 漏桶（Leaky Bucket）

请求先入桶，以固定速率流出，桶满则丢弃/排队。输出速率恒定，平滑尖峰，但无法容忍突发。

3.3 削峰填谷：MQ 的核心价值

大促/秒杀场景的经典做法是用 MQ 削峰：

流量高峰：10万 QPS ──► 网关限流 ──► Kafka（削峰蓄洪）──► 下游按 5000 QPS 匀速消费
                              ▲                          ▲
                    瞬时洪峰被缓冲              消费速率平稳，下游不被打爆

时间线（秒）：
  第 1s：涌入 10 万请求，全部投递到 Kafka
  第 2~20s：消费者以 5000/s 匀速消费，20s 内消化完
  下游数据库峰值压力从 10万 降为 5000

3.4 缓冲与丢弃策略

当队列/内存容量有限时，需要明确的丢弃与降级策略：

4. 生产实践

4.1 有界队列 + 拒绝策略

线程池 + 有界队列是进程内背压的基础。Java 线程池的拒绝策略就是在"背压的最后一公里"做决策：

ThreadPoolExecutor executor = new ThreadPoolExecutor(
    8,                    // 核心线程
    32,                   // 最大线程
    60, TimeUnit.SECONDS,
    new ArrayBlockingQueue<>(10_000),   // 有界队列，背压的"缓冲区"
    new ThreadPoolExecutor.AbortPolicy() {
        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {
            // 队列满 + 线程满：拒绝并反馈上游，触发上游退避
            throw new BusyException("queue full, apply backpressure");
        }
    });

关键：队列必须是有界的。无界队列会掩盖背压，最终 OOM 或延迟飙升。

4.2 Go channel 背压

Go 的 channel 天然是有界缓冲 + 阻塞语义：

const queueSize = 10_000
jobs := make(chan Job, queueSize)   // 有界缓冲

// 生产者：队列满时 send 阻塞，天然背压（生产者自动放慢）
go func() {
    for j := range jobSource {
        select {
        case jobs <- j:
            // 已入队
        case <-ctx.Done():
            return
        }
    }
}()

// 消费者：固定并发
for i := 0; i < 16; i++ {
    go worker(ctx, jobs)
}

当 jobs 缓冲满时，生产者 send 阻塞——这就是 Go 原生背压：上游自然放慢，而不是无限积压。

4.3 Kafka 消费者背压控制

# 消费者背压关键配置
max.poll.records=500        # 每轮最多拉取条数（越小背压越敏感）
max.poll.interval.ms=300000 # 两次 poll 最大间隔，超时会被认为死亡
fetch.max.bytes=5242880     # 单轮拉取字节上限

消费端要先落库/落盘再提交 offset，消费速度由下游存储能力决定，而不是由 Kafka 拉取能力决定：

consumer.poll(100).forEach(record -> {
    dbBatchWriter.write(record);      // 慢操作：写库
    consumer.commitSync();            // 写完再提交
});

4.4 背压与限流熔断的关系

背压、限流、熔断是稳定性体系的三个层面，常常配合使用：

生产者 ──► 限流（入口控流）──► 有界队列（背压缓冲）──► 消费者
                                              │
                                              ▼
                                        熔断（下游故障快速失败）

三者结合：入口限流挡住超容量流量，有界队列让背压可控，熔断在依赖故障时兜底。可结合 https://plumephp.com/rate-limiting-circuit-breaker/ 深入理解。

4.5 背压指标与告警

• 队列积压深度：队列长度 / 堆积消息数，超过阈值告警
• 消费 lag：Kafka 消费滞后量，滞后持续增长说明消费能力不足
• 拒绝率：被拒绝请求比例，过高说明容量规划不足
• 处理耗时趋势：消费者处理单条耗时上涨往往是积压的先行指标

# Prometheus 指标
async_queue_depth{queue="order-persist"}
kafka_consumer_lag{consumer_group="order-persist-group"}
async_rejected_total{queue="order-persist"}

5. 生产设计建议

1. 一切异步必须有界：队列、缓冲、线程池队列都要有上限，绝不使用无界队列
2. 明确背压策略：队列满时是阻塞、丢弃还是拒绝反馈？提前定义
3. 消费能力是锚点：削峰填谷时以"下游稳定消费速率"为准，而不是拍脑袋配并发
4. 积压要可视化：没有 lag/积压指标的异步系统等于盲飞
5. 重试也要背压：失败重试队列同样要有界，防止重试风暴放大压力
6. 关键链路降级：异步任务堆积时，允许降级为"同步兜底 + 对账"，保证核心可用

总结

异步化带来吞吐与解耦，背压控制则保证异步化不会失控。真正稳定的大规模异步系统，都是在"有界的缓冲 + 明确的拒绝/阻塞策略 + 可观测的积压指标"之上建立起来的。结合 https://plumephp.com/message-queue-deep-dive/ 与 https://plumephp.com/rate-limiting-circuit-breaker/，可以形成异步稳定性的完整图景。