用tokio的异步通道实现一个并发的哈希表

把tokio-mini-redis键值数据处理这部分抽象出来成了一个并发的哈希表，可以稍微加深一下redis的理解。 思路上就是把每一个get/insert之类的操作抽象成枚举命令（Command），每次解析完消息后，就向核心协程发送命令，这个所谓的核心协程是单线程处理数据的，同时也发送一个回调通道，在tokio-mini-redis里用的是tokio官方的tokio::sync::oneshot一次性通道的发送端，等待核心进行有序的返回结果。 对于一个简单mini-redis来说，不考虑其他权限功能、消息解析、淘汰策略，单单只是数据的crud，每秒可以承受远超过百万次的操作，作为内存级别的缓存是完全够用的。 本文意在抽象为一个并发的数据结构，所以用泛型简单封装了一下，其中为了避免繁杂的生命周期标记，泛型里去掉了引用而采用克隆的形式进行数据传递，所以键值都受到克隆Clone、跨线程传输Sync、静态生命周期'static的约束，另外核心部分是哈希表，所以键必然会受到哈希和比较Hash + Eq的约束。 当然，tokio-mini-redis直接使用了bytes::Bytes作为存储类型，也不需要考虑复杂的泛型约束以及生命周期标注的问题。 方法实现上，主要实现了insert、get、remove、len等基础方法，同时也实现了元素快照items的方法，基本上可以说，只要实现了以上五种方法，就可以实现几乎所有标准库里的其他方法了。 代码如下： // 命令枚举，理论上每个命令都会对应一个结果 // 和标准库hashmap不同的是，所有命令都避开了引用形式 enum Command<K, V> { Insert { key: K, value: V }, InsertResult { result: Option<V> }, Get { key: K }, GetResult { result: Option<V> }, Remove { key: K }, RemoveResult { result: Option<V> }, Items, ItemsResult { result: Vec<(K, V)> }, Len, LenResult { len: usize }, } // 引入tokio的无限容量异步通道和一次性通道 use tokio::{ self, sync::{ mpsc::{ unbounded_channel, UnboundedSender }, oneshot::{ channel, Sender }, } }; // 核心数据结构，以通道发送端的形式来存储基础数据结构，整体需要实现Clone特性 #[derive(Clone)] struct ChanHashMap<K, V> { sender: UnboundedSender<(Command<K, V>, Sender<Command<K, V>>)> } // 哈希trait是需要单独引入的，其他trait已经自动引入可以直接使用 use std::collections::HashMap; use std::hash::Hash; // 为核心数据结构撰写方法，为泛型键值提供必要的约束 impl<K, V> ChanHashMap<K, V> where K: Hash + Eq + Clone + Send + 'static, V: Clone + Send + 'static { fn new() -> Self { // 初始化单线程的哈希表作为核心存储数据结构 let mut database = HashMap::new(); // 初始化公共的输入输出异步通道，让其他命令可以通过此通道传输到主协程 let (in_sender, mut in_receiver) = unbounded_channel::<(Command<K, V>, Sender<Command<K, V>>)>(); // 启动一个协程（单线程）处理命令 tokio::spawn(async move { // 其他协程通过传输命令的形式传进while循环中进行单线程处理 while let Some((cmd, out_sender)) = in_receiver.recv().await { match cmd { // 进行命令枚举匹配，并通过out_sender发送回原本的协程 Command::Insert { key, value } => { let result = database.insert(key, value); let _ = out_sender.send(Command::InsertResult{ result }); } Command::Get { key } => { let result = match database.get(&key) { // std的hashmap.get获得是值引用，在并发情景下改为克隆传输，能够适用于绝大部分基础数据类型 Some(value) => Some(value.clone()), None => None, }; let _ = out_sender.send(Command::GetResult{ result }); } Command::Remove { key } => { let result = database.remove(&key); let _ = out_sender.send(Command::RemoveResult{ result }); } Command::Items => { // 通过快照的形式获取全部键值对，并存成数组返回 let result = database.iter().map(|(k, v)| (k.clone(), v.clone())).collect::<Vec<_>>(); let _ = out_sender.send(Command::ItemsResult{ result }); } Command::Len => { let _ = out_sender.send(Command::LenResult{ len: database.len() }); } _ => todo!() } } }); // 把初始化的通道发送端装入基础数据结构中，方便多线程状态下克隆拷贝 Self { sender: in_sender } } async fn insert(&mut self, key: K, value: V) -> Option<V> { // 初始化一次性通道 let (sender, receiver) = channel(); // 将命令和一次性通道的发送端发送到处理数据的主协程之中 let _ = self.sender.send((Command::Insert { key, value }, sender)); // 一次性通道的接收端等待处理结果并输出结果 match receiver.await.unwrap() { Command::InsertResult{ result } => result, _ => unreachable!() } } async fn get(&mut self, key: K) -> Option<V> { let (sender, receiver) = channel(); let _ = self.sender.send((Command::Get { key }, sender)); match receiver.await.unwrap() { Command::GetResult{ result } => result, _ => unreachable!() } } async fn remove(&mut self, key: K) -> Option<V> { let (sender, receiver) = channel(); let _ = self.sender.send((Command::Remove { key }, sender)); match receiver.await.unwrap() { Command::RemoveResult{ result } => result, _ => unreachable!() } } async fn items(&mut self) -> Vec<(K, V)> { let (sender, receiver) = channel(); let _ = self.sender.send((Command::Items , sender)); match receiver.await.unwrap() { Command::ItemsResult{ result } => result, _ => unreachable!() } } async fn len(&mut self) -> usize { let (sender, receiver) = channel(); let _ = self.sender.send((Command::Len, sender)); match receiver.await.unwrap() { Command::LenResult{ len } => len, _ => unreachable!() } } } 单机百万次操作大约在370ms-500ms这样，如果将异步通道换成crossbeam推荐的async-channel性能大约还能优化20-30%，不过一次性通道却没有更好的替代品，第三方异步库的性能基本都不如tokio自带的oneshot。 放在redis这个情境中，加上tcp-io以及RESP协议消息解析的性能损耗，最终每秒操作次数大约在10^5数量级也是比较符合认知的。