Rust多客户端TCP服务器消息转发异常：非发送端需主动触发接收

我正在实现一个支持多客户端的TCP服务器，需求是把一个客户端发送的消息转发给其他客户端，但遇到了问题：当某个客户端发消息时，服务器能正常接收，但其他客户端只有在自己给服务器发消息之后，才能收到之前的转发内容。使用的ThreadPool来自Rust官方书籍。

客户端代码 #

客户端模块 #

const BUFFER_SIZE: usize = 1024;

pub struct Client {
    write_stream: TcpStream,
    read_stream: TcpStream,
    pool: ThreadPool,
}

impl Client {
    pub fn new(config: Config) -> Self {
        let stream = TcpStream::connect(config.build_socket()).unwrap();
        Client {
            write_stream: stream.try_clone().unwrap(),
            read_stream: stream.try_clone().unwrap(),
            pool: ThreadPool::build(2),
        }
    }

    pub fn read_from_server(&self) {
        let mut buffer = [0; BUFFER_SIZE];
        let mut stream = self.read_stream.try_clone().unwrap();
        self.pool.execute(move || loop {
            match stream.read(&mut buffer) {
                Ok(n) => {
                    if n == 0 {
                        break;
                    }
                    let message = String::from_utf8_lossy(&buffer[..n]).to_string();
                    println!("Message from: {}", message);
                }
                Err(e) => {
                    eprintln!("Error while reading message: {}", e);
                    break;
                }
            };
        });
    }

    pub fn write_to_server(&self) -> Sender<String> {
        let mut stream = self.write_stream.try_clone().unwrap();
        let (tx, rx) = mpsc::channel::<String>();

        self.pool.execute(move || loop {
            let message = rx.recv().unwrap();
            if let Err(e) = stream.write(message.as_bytes()) {
                eprintln!("Error while sending message: {}", e);
            }
        });

        tx
    }
}

客户端主函数 #

fn main() {
    let config = Config::build(env::args());
    let client = Client::new(config);

    client.read_from_server();

    let tx = client.write_to_server();

    loop {
        let mut buffer = String::new();
        io::stdin().read_line(&mut buffer).unwrap();
        tx.send(buffer).unwrap();
    }
}

服务器代码 #

服务器模块 #

pub struct Server {
    pub listener: TcpListener,
    pub pool: ThreadPool,
    pub clients_count: usize,
    pub clients_sender: Arc<Mutex<Vec<(usize, Sender<String>)>>>,
}

impl Server {
    pub fn new(config: Config) -> Self {
        let listener = TcpListener::bind(config.build_socket()).unwrap();

        println!(
            "Creating server on address {} and port {}",
            config.ip_address, config.port
        );

        Server {
            listener,
            pool: ThreadPool::build(config.number_of_threads),
            clients_count: 0,
            clients_sender: Arc::new(Mutex::new(Vec::new())),
        }
    }

    pub fn listen(&mut self) {
        let mut id: usize = 0;

        for stream in self.listener.incoming() {
            match stream {
                Ok(s) => {
                    println!("{:?}", s.peer_addr());
                    let client = Client::new(s, id);
                    self.clients_count += 1;
                    let clients_sender = Arc::clone(&self.clients_sender);

                    self.pool.execute(move || {
                        let (tx, rx) = mpsc::channel::<String>();
                        let clone_senders = Arc::clone(&clients_sender);
                        clone_senders.lock().unwrap().push((id, tx.clone()));
                        client.handle_client(rx, clone_senders);
                    });

                    id += 1;
                }
                Err(_) => {
                    eprintln!("Error while parsing incoming stream");
                }
            }
        }
    }
}

客户端模块（服务器端） #

const SIZE_OF_BUFFER: usize = 1024;

#[derive(Debug)]
pub struct Client {
    pub stream: TcpStream,
    pub id: usize,
}

impl Client {
    pub fn new(stream: TcpStream, id: usize) -> Self {
        Client { stream, id }
    }

    pub fn handle_client(
        &mut self,
        rx: Receiver<String>,
        client_sender: Arc<Mutex<Vec<(usize, Sender<String>)>>>,
    ) {
        loop {
            let mut buffer = [0; SIZE_OF_BUFFER];
            let result = self.stream.read(&mut buffer);
            match result {
                Ok(_) => {
                    let message = String::from_utf8_lossy(&buffer).to_string();
                    let header = format!("<Client:{}>: ", self.id);
                    let payload = format!("{}{}", header, message);
                    println!("Message from client<{}>: {}", self.id, message);

                    client_sender
                        .lock()
                        .unwrap()
                        .iter()
                        .filter_map(|(id, tx)| {
                            if *id != self.id {
                                Some(Arc::new(tx.clone()))
                            } else {
                                None
                            }
                        })
                        .for_each(|tx| {
                            tx.send(payload.clone()).unwrap();
                        });
                }
                Err(_) => {
                    eprintln!("Error while reading from stream");
                    break;
                }
            }

            let mut write_stream = self.stream.try_clone().unwrap();

            for message in rx.try_iter() {
                write_stream.write_all(message.as_bytes()).unwrap();
            }
        }
    }
}

服务器主函数 #

fn main() {
    let config = Config::build(env::args());
    let mut server: Server = Server::new(config);

    server.listen();
}

问题分析与修复 #

问题根源 #

核心问题在服务器端Client的handle_client方法：

• 主循环被stream.read(&mut buffer)阻塞，这是同步阻塞调用，只有当前客户端发送数据时，才会执行后续的rx.try_iter()逻辑；
• rx.try_iter()是非阻塞的轮询，只会在read返回后才会检查通道中的消息，导致其他客户端的转发消息被积压在通道里，直到当前客户端主动发消息才会被处理。

修复方案 #

将读取客户端输入和处理转发消息的逻辑拆分为两个独立线程，避免互相阻塞：

1. 一个线程负责阻塞读取当前客户端的输入，广播消息给其他客户端；
2. 另一个线程负责阻塞接收转发消息，写入当前客户端的TCP流。

修改后的服务器端Client代码：

const SIZE_OF_BUFFER: usize = 1024;

#[derive(Debug)]
pub struct Client {
    pub stream: TcpStream,
    pub id: usize,
}

impl Client {
    pub fn new(stream: TcpStream, id: usize) -> Self {
        Client { stream, id }
    }

    pub fn handle_client(
        mut self,
        rx: Receiver<String>,
        client_sender: Arc<Mutex<Vec<(usize, Sender<String>)>>>,
    ) {
        // 克隆TCP流用于转发消息写入
        let write_stream = self.stream.try_clone().unwrap();
        
        // 线程1：处理当前客户端输入并广播
        std::thread::spawn(move || {
            loop {
                let mut buffer = [0; SIZE_OF_BUFFER];
                match self.stream.read(&mut buffer) {
                    Ok(n) => {
                        if n == 0 {
                            // 客户端断开连接，退出线程
                            break;
                        }
                        // 去除消息中的换行符和空白
                        let message = String::from_utf8_lossy(&buffer[..n]).trim().to_string();
                        let header = format!("<Client:{}>: ", self.id);
                        let payload = format!("{}{}\n", header, message);
                        println!("Message from client<{}>: {}", self.id, message);

                        // 广播给其他客户端
                        let senders = client_sender.lock().unwrap();
                        senders.iter()
                            .filter(|(id, _)| *id != self.id)
                            .for_each(|(_, tx)| {
                                // 忽略通道断开的错误，避免panic
                                let _ = tx.send(payload.clone());
                            });
                    }
                    Err(e) => {
                        eprintln!("Error reading from client {}: {}", self.id, e);
                        break;
                    }
                }
            }
            // 客户端断开后，从全局sender列表移除自身
            let mut senders = client_sender.lock().unwrap();
            senders.retain(|(id, _)| *id != self.id);
        });

        // 线程2：接收转发消息并写入客户端
        std::thread::spawn(move || {
            for message in rx {
                if let Err(e) = write_stream.write_all(message.as_bytes()) {
                    eprintln!("Error sending to client: {}", e);
                    break;
                }
            }
        });
    }
}

额外优化点 #

• 将handle_client的参数从&mut self改为self，确保线程能获取客户端的所有权；
• 处理read返回0的情况（客户端断开），并从全局sender列表中清理无效的客户端；
• 使用trim()处理输入消息，避免多余的换行符；
• 发送消息时忽略通道错误，防止因单个客户端断开导致广播逻辑panic；
• 用rx的阻塞迭代替代try_iter()，确保消息能被及时处理。

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