Rust async get_or_insert_with conncurrent map 封装（基于dashmap）

飞书用户6188

2月10日修改

AsyncInitConcurrentMap 封装（基于dashmap）

设计动机

在 Rust 异步编程中，我们经常遇到这样的场景：需要一个并发 Map，同时要支持异步初始化。​

现有方案的问题

1.
DashMap 的 or_insert 限制：​

代码块

// DashMap 的 or_insert_with 是同步的​
map.entry(key).or_insert_with(|| {// 这里不能直接用 async 函数// 如果在这里调用 block_on 会导致严重问题​
});​

2.
同步调用异步的问题：​
◦
如果在同步上下文中调用异步函数（如使用 block_on）​
◦
当前线程会被阻塞​
◦
导致其他异步任务无法调度​
◦
可能引发死锁​

解决方案

我们的方案是将异步初始化逻辑从 entry 的回调中分离出来：

代码块

// 不在 or_insert_with 中执行异步初始化let entry = map.entry(key).or_insert_with(|| {// 只创建初始状态​
    ValueState::Initializing(tx)​
});​
​
// 在单独的异步任务中执行初始化​
tokio::spawn(async move {// 这里可以安全地执行异步操作match init_fut.await {Ok(value) => {let _ = tx.send(value.clone());  // 先发送值​
            inner.insert(key, ValueState::Ready(value));  // 再更新状态​
        }Err(e) => {​
            inner.remove(&key);drop(tx);  // 通知错误​
        }​
    }​
});​

核心实现

状态管理

设计原因：

•
使用枚举保证状态转换的类型安全​

•
将通知 channel 绑定到初始化状态，确保生命周期正确​

•
避免使用额外的标志位，保持内存效率​

代码块

enum ValueState<V> {Initializing(broadcast::Sender<V>),  // channel 直接传递值Ready(V),​
}​

关键细节：

•
Initializing 持有 broadcast::Sender 而不是 oneshot，支持多个等待者​

•
Ready 直接持有值，避免额外的引用计数​

•
枚举设计使得状态检查在编译时完成​

读写分离设计

设计原因：

•
读操作应该尽可能快速且无阻塞​

•
写操作需要保证原子性，但要最小化锁持有时间​

•
异步等待不能持有任何锁​

1.
快速路径（读）：​

代码块

if let Some(entry) = self.inner.get(&key) {  // 只获取读锁match entry.value() {​
        ValueState::Ready(v) => return Ok(v.clone()),​
        ValueState::Initializing(tx) => {let mut rx = tx.subscribe();drop(entry);  // 立即释放读锁return Ok(rx.recv().await?);​
        }​
    }​
}​

Rust async get_or_insert_with conncurrent map 封装（基于dashmap）​

Rust async get_or_insert_with conncurrent map 封装（基于dashmap）