Rust+Tokio存储TcpStream至HashMap时遭遇死锁问题排查

问题背景 #

开发简易TCP广播程序，需求如下：

• 新客户端连接时，向所有已连接客户端广播上线消息
• 接收客户端消息并广播至其他所有客户端

使用nc localhost 8080测试时出现死锁：注释掉Server::handle_client_events(clients_clone, id).await;后，广播功能正常；启用该行后，第一个客户端连接后，后续新客户端无法正常工作。

死锁核心原因 #

1. Mutex锁长时间持有：handle_client_events中获取客户端列表的Mutex锁后，直接执行socket.read(&mut buf).await——异步read会等待客户端输入，期间锁一直被占用，导致主循环的客户端插入操作、其他广播操作都无法获取锁，彻底阻塞。
2. 变量引用错误：handle_connection的异步任务中使用了未定义的clients_clone，属于编译级错误。
3. 主循环被不必要阻塞：start函数中await Server::handle_connection(...)会等待该函数执行完毕，而handle_connection仅用于创建异步任务，此处await会导致主循环无法及时处理新连接。
4. Socket并发访问冲突：客户端列表直接存储TcpStream，而TcpStream不支持并发读写，且整个列表被锁会导致所有IO操作串行化，效率极低且易引发阻塞。

修复方案与代码实现 #

关键修复点 #

• 将客户端存储结构改为HashMap<usize, Arc<Mutex<TcpStream>>>，让每个Socket可独立锁定，避免整个列表被长时间占用
• 所有异步IO操作（read/write）均在释放Mutex锁后执行，仅在修改客户端列表或获取Socket引用时短暂持有锁
• 修正变量引用错误，移除主循环中不必要的await
• 优化广播逻辑：先复制客户端Socket引用列表，再逐个发送消息，避免长时间持有客户端列表锁

修复后完整代码 #

use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::Mutex;
use tokio::{net::TcpListener, net::TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[derive(Debug)]
struct Server {
    clients: Arc<Mutex<HashMap<usize, Arc<Mutex<TcpStream>>>>>,
}

impl Server {
    async fn start(&self, addr: &str) -> tokio::io::Result<()> {
        let server = TcpListener::bind(&addr).await?;
        println!("Server listening on {}", addr);

        let mut id: usize = 0;
        let mut num_clients: i16 = 0;
        loop {
            let (socket, _) = server.accept().await?;
            num_clients += 1;
            id += 1;
            println!("Clients online: {}", num_clients);

            // 插入新客户端，仅短暂持有锁
            {
                let mut clients = self.clients.lock().await;
                clients.insert(id, Arc::new(Mutex::new(socket)));
            }

            // 异步处理连接，不阻塞主循环
            let clients_clone = self.clients.clone();
            tokio::spawn(async move {
                Server::handle_connection(clients_clone, id).await;
            });
        }
    }

    async fn handle_client_events(
        client_socket: Arc<Mutex<TcpStream>>,
        client_id: usize,
        clients: Arc<Mutex<HashMap<usize, Arc<Mutex<TcpStream>>>>>
    ) {
        let mut buf = vec![0; 1024];
        loop {
            // 仅锁定当前客户端的Socket
            let mut socket = client_socket.lock().await;
            match socket.read(&mut buf).await {
                Ok(0) => {
                    println!("Client {} disconnected", client_id);
                    // 移除客户端时锁定整个列表
                    let mut clients = clients.lock().await;
                    clients.remove(&client_id);
                    break;
                }
                Ok(n) => {
                    let msg = format!("[Client {}]: {}", client_id, String::from_utf8_lossy(&buf[..n]));
                    // 释放当前Socket锁后执行广播
                    drop(socket);
                    Server::stream_to_all(&clients, msg).await;
                    // 重置缓冲区
                    buf.fill(0);
                }
                Err(e) => {
                    println!("Socket error for client {}: {}", client_id, e);
                    let mut clients = clients.lock().await;
                    clients.remove(&client_id);
                    break;
                }
            }
        }
    }

    async fn handle_connection(clients: Arc<Mutex<HashMap<usize, Arc<Mutex<TcpStream>>>>>, id: usize) {
        let msg = format!("Client id {} connected\n", id);
        // 广播上线消息
        Server::stream_to_all(&clients, msg).await;

        // 获取当前客户端的Socket引用，短暂持有锁
        let client_socket = {
            let clients = clients.lock().await;
            clients.get(&id).unwrap().clone()
        };

        // 处理客户端消息
        Server::handle_client_events(client_socket, id, clients).await;
    }

    async fn stream_to_all(clients: &Arc<Mutex<HashMap<usize, Arc<Mutex<TcpStream>>>>>, message: String) {
        // 先复制所有Socket引用，立即释放列表锁
        let clients_list = clients.lock().await.values().cloned().collect::<Vec<_>>();
        drop(clients);

        // 逐个发送消息
        for client_socket in clients_list {
            let mut socket = client_socket.lock().await;
            if let Err(e) = socket.write_all(message.as_bytes()).await {
                println!("Could not send message to client: {}", e);
            }
        }
    }
}

#[tokio::main]
async fn main() -> tokio::io::Result<()> {
    let addr = "127.0.0.1:8080";
    let server = Server {
        clients: Arc::new(Mutex::new(HashMap::new())),
    };
    server.start(addr).await?;
    Ok(())
}

验证效果 #

• 第一个客户端连接：收到Client id 1 connected
• 第二个客户端连接：第一个客户端收到Client id 2 connected，第二个客户端收到Client id 2 connected
• 任意客户端发送消息：其他所有客户端都会收到带客户端ID的格式化消息

内容的提问来源于stack exchange，提问作者Oliver Weitman