Rust全双工TCP服务器/客户端代码故障排查与最佳实践

问题描述 #

我尝试用Rust实现全双工TCP服务器和客户端，但当前代码无法正常收发数据，服务器退出后客户端还会陷入无限循环。需要明确修改点以及全双工TCP的实现最佳实践。

原始代码 #

client.rs #

use std::io::{self, Read, Write};
use std::net::TcpStream;
use std::thread;

static mut FLAG: bool = false;

fn send_to_server(mut stream: TcpStream) {
    unsafe {
        while !FLAG {
            let mut send_msg = String::new();
            if let Ok(_) = io::stdin().read_line(&mut send_msg) {
                stream
                    .write_all(send_msg.as_bytes())
                    .expect("Failed to send message to server");
            }
        }
    }
}

fn recv_from_server(mut stream: TcpStream) {
    unsafe {
        loop {
            let mut data = String::new();
            if let Ok(_) = stream.read_to_string(&mut data) {
                if data.trim() == "quit" {
                    println!("Closing connection");
                    FLAG = true;
                    break;
                }
                println!("Server: {}", data);
            }
        }
    }
}

fn main() {
    let mut threads = vec![];

    let host = "localhost";
    let port = 6500;

    if let Ok(mut client_socket) = TcpStream::connect(format!("{}:{}", host, port)) {
        println!("Client is connected to the Server");

        if let Ok(send_socket) = client_socket.try_clone() {
            let t_send = thread::spawn(move || send_to_server(send_socket));
            threads.push(t_send);
        }

        let recv_socket = client_socket;
        let t_rcv = thread::spawn(move || recv_from_server(recv_socket));
        threads.push(t_rcv);

        for thread in threads {
            thread.join().expect("Thread panicked");
        }
    } else {
        eprintln!("Failed to connect to server");
    }

    println!("EXITING");
}

server.rs #

use std::io::{self, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::thread;

static mut FLAG: bool = false;

fn recv_from_client(mut conn: TcpStream) {
    unsafe {
        loop {
            if FLAG {
                break;
            }
            let mut message = String::new();
            if let Ok(_) = conn.read_to_string(&mut message) {
                if message.trim() == "quit" {
                    conn.write_all(b"quit")
                        .expect("Failed to send 'quit' message");
                    conn.shutdown(std::net::Shutdown::Both)
                        .expect("Failed to shutdown connection");
                    println!("Connection Closed");
                    FLAG = true;
                    break;
                }
                println!("Client: {}", message);
            }
        }
    }
}

fn send_to_client(mut conn: TcpStream) {
    unsafe {
        loop {
            if FLAG {
                break;
            }
            let mut send_msg = String::new();
            if let Ok(_) = io::stdin().read_line(&mut send_msg) {
                if send_msg.trim() == "quit" {
                    conn.write_all(b"quit")
                        .expect("Failed to send 'quit' message");
                    conn.shutdown(std::net::Shutdown::Both)
                        .expect("Failed to shutdown connection");
                    println!("Connection Closed");
                    FLAG = true;
                    break;
                }
                conn.write_all(send_msg.as_bytes())
                    .expect("Failed to send message to client");
            }
        }
    }
}

fn main() {
    let mut threads = vec![];

    let host = "localhost";
    let server_port = 6500;

    if let Ok(server_socket) = TcpListener::bind(format!("{}:{}", host, server_port)) {
        println!("Socket binded.");

        println!("Listening.....");
        if let Ok((connection_socket, addr)) = server_socket.accept() {
            println!("Connection Established with a Client on {}", addr);

            if let Ok(recv_socket) = connection_socket.try_clone() {
                let t_rcv = thread::spawn(move || recv_from_client(recv_socket));
                threads.push(t_rcv);
            }

            if let Ok(send_socket) = connection_socket.try_clone() {
                let t_send = thread::spawn(move || send_to_client(send_socket));
                threads.push(t_send);
            }
        } else {
            eprintln!("Failed to establish connection with client");
            return;
        }

        for thread in threads {
            thread.join().expect("Thread panicked");
        }

        println!("EXITING");
    } else {
        eprintln!("Failed to bind socket");
    }
}

核心问题与修复步骤 #

1. 替换read_to_string为按行读取 #

read_to_string会持续阻塞直到TCP流触发EOF（连接关闭），导致正常收发消息时线程无法响应。改用BufReader::read_line，因为代码中用read_line输入消息（带换行符），可以每次读取完整的一行消息。

2. 移除不安全的全局静态变量FLAG #

全局unsafe静态变量会引发数据竞争，行为不可预测。改用线程安全的Arc<AtomicBool>传递退出信号，确保多线程间状态同步。

3. 正确处理IO操作返回值 #

• 检查read_line的返回值：返回Ok(0)表示连接已关闭，需退出循环
• 捕获IO错误并优雅退出，避免程序直接panic

4. 修复客户端无限循环问题 #

客户端接收线程在服务器关闭连接后，需通过AtomicBool通知发送线程停止运行，避免无限阻塞。

修改后的代码示例 #

client.rs #

use std::io::{self, BufRead, BufReader, Write};
use std::net::TcpStream;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

fn send_to_server(mut stream: TcpStream, running: Arc<AtomicBool>) {
    while running.load(Ordering::Relaxed) {
        let mut send_msg = String::new();
        match io::stdin().read_line(&mut send_msg) {
            Ok(_) => {
                if send_msg.trim() == "quit" {
                    let _ = stream.write_all(b"quit\n");
                    running.store(false, Ordering::Relaxed);
                    break;
                }
                if let Err(e) = stream.write_all(send_msg.as_bytes()) {
                    eprintln!("Send failed: {}", e);
                    running.store(false, Ordering::Relaxed);
                    break;
                }
            }
            Err(e) => {
                eprintln!("Read stdin failed: {}", e);
                running.store(false, Ordering::Relaxed);
                break;
            }
        }
    }
}

fn recv_from_server(mut stream: TcpStream, running: Arc<AtomicBool>) {
    let reader = BufReader::new(&stream);
    for line in reader.lines() {
        match line {
            Ok(data) => {
                if data.trim() == "quit" {
                    println!("Closing connection");
                    running.store(false, Ordering::Relaxed);
                    break;
                }
                println!("Server: {}", data);
            }
            Err(e) => {
                eprintln!("Receive failed: {}", e);
                running.store(false, Ordering::Relaxed);
                break;
            }
        }
        if !running.load(Ordering::Relaxed) {
            break;
        }
    }
    running.store(false, Ordering::Relaxed);
}

fn main() {
    let host = "localhost";
    let port = 6500;

    if let Ok(client_socket) = TcpStream::connect(format!("{}:{}", host, port)) {
        println!("Client is connected to the Server");

        let running = Arc::new(AtomicBool::new(true));

        let send_socket = client_socket.try_clone().expect("Clone stream failed");
        let send_running = Arc::clone(&running);
        let t_send = thread::spawn(move || send_to_server(send_socket, send_running));

        let recv_running = Arc::clone(&running);
        let t_rcv = thread::spawn(move || recv_from_server(client_socket, recv_running));

        t_send.join().expect("Send thread panicked");
        t_rcv.join().expect("Receive thread panicked");
    } else {
        eprintln!("Failed to connect to server");
    }

    println!("EXITING");
}

server.rs #

use std::io::{self, BufRead, BufReader, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;

fn recv_from_client(mut stream: TcpStream, running: Arc<AtomicBool>) {
    let reader = BufReader::new(&stream);
    for line in reader.lines() {
        match line {
            Ok(message) => {
                if message.trim() == "quit" {
                    let _ = stream.write_all(b"quit\n");
                    let _ = stream.shutdown(std::net::Shutdown::Both);
                    println!("Connection Closed");
                    running.store(false, Ordering::Relaxed);
                    break;
                }
                println!("Client: {}", message);
            }
            Err(e) => {
                eprintln!("Receive failed: {}", e);
                running.store(false, Ordering::Relaxed);
                break;
            }
        }
        if !running.load(Ordering::Relaxed) {
            break;
        }
    }
    running.store(false, Ordering::Relaxed);
}

fn send_to_client(mut stream: TcpStream, running: Arc<AtomicBool>) {
    while running.load(Ordering::Relaxed) {
        let mut send_msg = String::new();
        match io::stdin().read_line(&mut send_msg) {
            Ok(_) => {
                if send_msg.trim() == "quit" {
                    let _ = stream.write_all(b"quit\n");
                    let _ = stream.shutdown(std::net::Shutdown::Both);
                    println!("Connection Closed");
                    running.store(false, Ordering::Relaxed);
                    break;
                }
                if let Err(e) = stream.write_all(send_msg.as_bytes()) {
                    eprintln!("Send failed: {}", e);
                    running.store(false, Ordering::Relaxed);
                    break;
                }
            }
            Err(e) => {
                eprintln!("Read stdin failed: {}", e);
                running.store(false, Ordering::Relaxed);
                break;
            }
        }
    }
}

fn main() {
    let host = "localhost";
    let server_port = 6500;

    if let Ok(server_socket) = TcpListener::bind(format!("{}:{}", host, server_port)) {
        println!("Socket binded.");
        println!("Listening.....");

        if let Ok((connection_socket, addr)) = server_socket.accept() {
            println!("Connection Established with a Client on {}", addr);

            let running = Arc::new(AtomicBool::new(true));

            let recv_socket = connection_socket.try_clone().expect("Clone stream failed");
            let recv_running = Arc::clone(&running);
            let t_rcv = thread::spawn(move || recv_from_client(recv_socket, recv_running));

            let send_socket = connection_socket.try_clone().expect("Clone stream failed");
            let send_running = Arc::clone(&running);
            let t_send = thread::spawn(move || send_to_client(send_socket, send_running));

            t_rcv.join().expect("Receive thread panicked");
            t_send.join().expect("Send thread panicked");
        } else {
            eprintln!("Failed to establish connection with client");
            return;
        }

        println!("EXITING");
    } else {
        eprintln!("Failed to bind socket");
    }
}

全双工TCP实现最佳实践 #

1. 优先使用异步IO框架
   高并发场景下，Tokio、async-std等异步框架比线程模型更高效，无需为每个连接创建独立线程，减少资源开销。

2. 使用线程安全的同步机制
   绝不使用unsafe全局变量做线程通信，优先选择：

   • 原子类型：适用于简单状态标记
   • MPSC通道：适用于复杂消息传递
   • Mutex/RwLock：适用于共享复杂数据结构的场景

3. 明确消息边界
   TCP是字节流协议，需自定义消息分隔规则：

   • 文本协议用换行符\n分隔
   • 二进制协议用长度前缀（先发送消息长度，再发送内容）
     避免依赖EOF的读取方法（如read_to_string）。

4. 优雅处理连接关闭与错误

   • 检查read返回的Ok(0)，识别连接关闭
   • 捕获IO错误并优雅退出，避免panic
   • 调用shutdown明确关闭连接的读写方向，确保对方收到关闭信号

5. 设置合理超时
   通过TcpStream::set_read_timeout/set_write_timeout设置超时，避免线程因网络问题无限阻塞。

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