如何在Rust终端TCP聊天程序中仅在连接存活时读取stdin？

核心问题分析 #

默认的std::io::stdin().read_line()是阻塞式同步IO，它会一直等待用户输入直到换行，期间无法感知TCP连接的断开状态——只有当你发送消息时才会发现连接已死。要解决这个问题，必须让程序同时监听两个事件：用户输入和连接状态变化，这在Rust中最惯用的方式是使用异步运行时（比如Tokio）结合事件驱动模型。

方案1：模仿SSH行为（基础版） #

当连接断开时，立即中止stdin读取，打印连接丢失提示，这是最容易实现的版本，核心是用tokio::select!宏同时等待stdin输入和连接的异常信号。

依赖准备 #

在Cargo.toml中添加：

tokio = { version = "1.0", features = ["full"] }

实现代码 #

use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let mut stream = TcpStream::connect("127.0.0.1:8080").await?;
    let (read_half, mut write_half) = stream.split();
    let mut stdin = BufReader::new(tokio::io::stdin()).lines();

    // 启动后台任务监听TCP连接的读取（连接断开时会返回错误）
    let mut connection_watch = tokio::spawn(async move {
        let mut buf = [0; 1024];
        loop {
            match read_half.read(&mut buf).await {
                Ok(0) => break, // 对方主动关闭连接
                Err(_) => break, // 连接异常中断
                Ok(_) => { /* 处理收到的消息，此处省略逻辑 */ },
            }
        }
        Err("连接已丢失".into())
    });

    loop {
        tokio::select! {
            // 等待用户输入
            input = stdin.next_line() => {
                let input = input?.unwrap_or_default();
                if input.is_empty() { continue; }
                // 尝试发送消息
                if let Err(e) = write_half.write_all(input.as_bytes()).await {
                    eprintln!("发送失败：{}", e);
                    break;
                }
            },
            // 连接异常触发
            res = &mut connection_watch => {
                match res {
                    Ok(Err(msg)) => eprintln!("\n{}", msg),
                    Err(_) => eprintln!("\n连接监听任务异常"),
                    _ => (),
                }
                break;
            }
        }
    }

    Ok(())
}

关键细节 #

• tokio::select!会同时等待多个异步任务，任意一个完成就执行对应分支，连接断开时能立刻打断stdin的等待。
• 后台任务监听TCP读取端：连接断开时read会返回错误或0（对方关闭），任务结束后触发select!的连接异常分支。
• 打印提示时用\n确保换行，避免覆盖用户正在输入的内容。

方案2：理想版（不中断用户输入+自动重连） #

要实现连接断开时不丢失用户正在输入的内容，需要接管终端的输入处理（禁用系统行缓冲），自己维护输入缓冲区，同时监听连接状态和用户按键。这里用crossterm库处理终端原始模式。

依赖准备 #

tokio = { version = "1.0", features = ["full"] }
crossterm = "0.27"

实现代码 #

use crossterm::{
    event::{self, Event, KeyCode, KeyEvent},
    terminal::{self, EnterAlternateScreen, LeaveAlternateScreen},
    ExecutableCommand,
};
use tokio::net::TcpStream;
use tokio::sync::mpsc;
use std::error::Error;
use std::io::{stdout, Write};

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // 进入终端原始模式，禁用系统行缓冲
    stdout().execute(EnterAlternateScreen)?;
    terminal::enable_raw_mode()?;

    // 用通道传递用户输入的字符，解耦按键捕获与主逻辑
    let (tx, mut rx) = mpsc::unbounded_channel();

    // 启动后台任务捕获用户按键
    tokio::spawn(async move {
        loop {
            if let Ok(Event::Key(KeyEvent { code, .. })) = event::read() {
                match code {
                    KeyCode::Char(c) => { tx.send(c).unwrap(); },
                    KeyCode::Backspace => { tx.send('\x08').unwrap(); }, // 退格键
                    KeyCode::Enter => { tx.send('\n').unwrap(); },
                    KeyCode::Esc => break, // 按ESC退出程序
                    _ => (),
                }
            }
        }
    });

    let mut input_buf = String::new();
    let mut stream: Option<TcpStream> = Some(TcpStream::connect("127.0.0.1:8080").await?);

    print!("> ");
    stdout().flush()?;

    loop {
        tokio::select! {
            // 处理用户按键
            Some(c) = rx.recv() => {
                match c {
                    '\n' => {
                        // 尝试发送消息
                        if let Some(ref mut s) = stream {
                            if let Err(e) = s.write_all(input_buf.as_bytes()).await {
                                eprintln!("\n连接已丢失，正在尝试重连...");
                                stream = None;
                            } else {
                                println!("\n已发送：{}", input_buf);
                            }
                        } else {
                            eprintln!("\n当前无连接，消息暂存：{}", input_buf);
                        }
                        input_buf.clear();
                        print!("> ");
                        stdout().flush()?;
                    },
                    '\x08' => {
                        // 退格删除字符并更新终端显示
                        if !input_buf.is_empty() {
                            input_buf.pop();
                            print!("\x08 \x08");
                            stdout().flush()?;
                        }
                    },
                    c => {
                        input_buf.push(c);
                        print!("{}", c);
                        stdout().flush()?;
                    }
                }
            },
            // 后台自动重连逻辑（仅当连接断开时执行）
            _ = async {
                while stream.is_none() {
                    match TcpStream::connect("127.0.0.1:8080").await {
                        Ok(s) => {
                            eprintln!("\n重连成功！");
                            stream = Some(s);
                            print!("> ");
                            stdout().flush()?;
                            break;
                        },
                        Err(_) => {
                            tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
                        }
                    }
                }
                Ok(())
            } => {},
        }
    }

    // 程序退出前恢复终端正常模式
    terminal::disable_raw_mode()?;
    stdout().execute(LeaveAlternateScreen)?;
    Ok(())
}

关键细节 #

• 终端原始模式：禁用系统行缓冲，让程序直接捕获每一个按键，自己维护输入缓冲区，避免连接断开时丢失未发送的输入内容。
• mpsc通道：将按键捕获任务与主逻辑解耦，主逻辑通过通道异步接收按键事件。
• 重连逻辑：连接断开后自动尝试重连，重连成功后恢复发送能力，用户正在输入的内容全程保留。

注意事项 #

• 错误处理：示例中用unwrap简化代码，实际项目中应使用?或match处理错误，避免程序panic。
• 终端恢复：必须确保程序退出时恢复终端模式，否则终端会处于异常状态（可通过drop或tokio::scope保证）。
• 跨平台：crossterm支持Windows、Linux、macOS，无需额外适配。

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