请求合并

概念定义

缓存 miss 总是成批到达。不合并时每个等待者独立回源，源站把同一查询执行 N 遍；合并推选一个 leader 去抓取，follower 订阅 leader 的结果，N 次相同上游调用折叠为一次。

工作机制

进程内：键到在途 promise 的映射，新到者加入既有 promise，完成即移除。代理层：nginx proxy_cache_lock 一类的缓存锁按 key 串行化填充尝试，超时则释放等待者自行回源。跨进程合并依赖分布式锁或基于租约的选举。

使用场景

到期或部署后的热键、昂贵聚合、一切缓存填充路径。它与 TTL 抖动互补：抖动摊开到期，合并吸收仍对齐的部分。

常见误区

合并是“并发去重”不是“时间窗批处理”。键必须承载全部请求语义——按 URL 合并却忽略认证或 Vary 头会张冠李戴。leader 停滞需要有限等待加回退自取；错误扇出要决策：全体等待者收到失败，还是新 leader 立刻重试并重建突发。非幂等操作绝不因“同时到达”就合并。

代码示例

export class SingleFlight {
  private readonly inFlight = new Map<string, Promise<unknown>>();

  async run<T>(key: string, loader: () => Promise<T>): Promise<T> {
    if (!key) throw new TypeError("key must be a non-empty string");
    const existing = this.inFlight.get(key);
    if (existing) return existing as Promise<T>;
    const started = (async () => {
      try {
        return await loader();
      } finally {
        this.inFlight.delete(key);
      }
    })();
    this.inFlight.set(key, started);
    return started;
  }

  pending(key: string): boolean {
    return this.inFlight.has(key);
  }
}

use std::collections::HashMap;
use std::future::Future;
use std::sync::Arc;
use tokio::sync::{broadcast, Mutex};

type SharedResult = Arc<Result<String, String>>;

#[derive(Clone, Default)]
struct SingleFlight {
    inflight: Arc<Mutex<HashMap<String, broadcast::Sender<SharedResult>>>>,
}

impl SingleFlight {
    fn new() -> Self {
        Self::default()
    }

    async fn run<F>(&self, key: &str, load: F) -> Result<String, String>
    where
        F: Future<Output = Result<String, String>>,
    {
        let follower = {
            let mut inflight = self.inflight.lock().await;
            match inflight.get(key) {
                Some(sender) => Some(sender.subscribe()),
                None => {
                    let (sender, _) = broadcast::channel(1);
                    inflight.insert(key.to_string(), sender);
                    None
                }
            }
        };

        if let Some(mut receiver) = follower {
            return match receiver.recv().await {
                Ok(shared) => (*shared).clone(),
                Err(_) => Err("leader cancelled before publishing".to_string()),
            };
        }

        let result = load.await;
        let shared = Arc::new(result.clone());
        let sender = {
            let mut inflight = self.inflight.lock().await;
            inflight.remove(key)
        };
        if let Some(sender) = sender {
            let _ = sender.send(shared);
        }
        result
    }
}

#[tokio::main]
async fn main() {
    let group = SingleFlight::new();
    let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let (first, second) = tokio::join!(
        group.run("user:1", async {
            calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            tokio::time::sleep(std::time::Duration::from_millis(30)).await;
            Ok::<String, String>("payload".to_string())
        }),
        group.run("user:1", async {
            calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            tokio::time::sleep(std::time::Duration::from_millis(30)).await;
            Ok::<String, String>("payload".to_string())
        })
    );
    println!(
        "{:?} {:?} loads={}",
        first,
        second,
        calls.load(std::sync::atomic::Ordering::SeqCst)
    );
}