异步 Singleflight 请求合并

功能说明

缓存未命中时，多个并发请求可能同时查询同一个后端资源，造成重复计算和连接压力。Singleflight 把相同 key 的并发调用合并为一次：第一个调用成为领导者并执行加载，其余调用成为跟随者，等待领导者完成后共享结果。成功和失败都会共享，避免失败瞬间所有请求再次同时重试。

本例用 tokio 的广播通道实现结果分发。领导者把结果放入 Arc 后广播，跟随者收到同一份结果并克隆返回。加载期间 key 会保存在共享映射中，领导者完成后移除，后续新请求可以重新发起加载。main 并发两次相同 key 的调用，用计数器证明加载只发生一次。

关键代码

inflight 映射的值是广播发送器，通道容量为一，只需要传递一次最终结果。run 先在异步互斥锁内查找 key：存在就订阅通道并作为跟随者，不存在就创建通道并插入，作为领导者。跟随者等待 recv，成功时解引用 Arc 并克隆内部 Result；发送器被丢弃而未发送时返回取消错误。领导者执行传入的 future，包装结果，再在锁内移除 key，释放锁后发送，避免持锁唤醒造成阻塞。

使用注意

结果类型被固定为字符串，实际使用可泛化为可克隆的 T，但要注意错误也必须可克隆。广播容量为一且只发送一次，若跟随者在发送之后才订阅会错过结果，因此订阅必须在领导者移除 key 之前完成，本例通过同一把锁保证顺序。领导者被取消时跟随者会收到关闭错误，应允许它们重试。不同 key 互不影响，但共享映射会随并发 key 数量增长。不要合并有副作用或不可幂等的写操作。

验证方式

运行脚本，两次调用返回相同的 payload，最后的 loads=1，说明只有一个 future 被真正轮询并执行加载。把两个 key 改成不同值，loads 应为 2，说明合并只针对相同 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)
    );
}