为什么选择 Rust 和 Tokio？
在系统编程领域，Rust 以其内存安全和性能优势脱颖而出。当我们需要构建高并发的网络服务时，异步编程成为不二之选。Tokio 是 Rust 生态中最成熟的异步运行时，提供了事件驱动、非阻塞 I/O 和任务调度的核心功能。
本文将通过一个完整的 TCP 聊天服务器示例，带你掌握 Tokio 的核心概念。你将学会如何处理并发连接、在不同任务间通信，以及如何优雅地关闭服务。
环境准备
确保你已经安装了 Rust（推荐使用 rustup 安装）。检查版本：
rustc --version
cargo --version
创建新项目：
cargo new tokio-chat
cd tokio-chat
在 Cargo.toml 中添加依赖：
[dependencies]
tokio = { version = "1.37", features = ["full"] }
full 特性会启用所有 Tokio 组件（例如 io、time、sync 等），方便开发。
异步基础
Rust 的异步函数使用 async fn 定义，调用时返回一个 Future，需要执行器（Executor）来驱动。Tokio 提供了两种运行时：多线程的 currentthread 和 multithread，默认是后者。
#[tokio::main]
async fn main() {
println!("Hello from Tokio");
}
#[tokio::main] 宏将异步 main 函数包装成运行时初始化和执行。
实战：构建 TCP 聊天服务器
我们将实现一个简单的聊天室，支持多个客户端连接，每个客户端发送的消息会广播给所有其他客户端。
第一步：接受连接
首先，我们监听一个 TCP 端口，并为每个连接生成一个任务。
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use std::sync::{Arc, Mutex};
use tokio::sync::broadcast;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:8080").await?;
println!("Server listening on port 8080");
// 创建广播通道，用于向所有客户端发送消息
let (tx, _rx) = broadcast::channel(10);
loop {
let (socket, addr) = listener.accept().await?;
println!("New connection from {}", addr);
// 为每个连接克隆一个发送端
let tx = tx.clone();
// 启动一个新任务处理该连接
tokio::spawn(async move {
if let Err(e) = handle_connection(socket, tx).await {
eprintln!("Error handling connection {}: {}", addr, e);
}
});
}
}
第二步：处理客户端消息
每个客户端连接我们读取其发送的每一行，然后广播给所有其他客户端。
async fn handle_connection(
mut socket: TcpStream,
tx: broadcast::Sender<String>,
) -> Result<(), Box<dyn std::error::Error>> {
// 将 socket 分成读和写部分
let (reader, mut writer) = socket.split();
let mut reader = BufReader::new(reader);
let mut line = String::new();
loop {
// 读取一行
match reader.read_line(&mut line).await {
Ok(0) => break, // 连接关闭
Ok(_) => {
// 发送到广播通道
tx.send(line.trim().to_string()).unwrap();
line.clear();
}
Err(e) => {
eprintln!("Error reading: {}", e);
break;
}
}
}
Ok(())
}
但是，我们还需要一个任务来接收广播的消息并发送给客户端。由于我们只有一个 writer，不能同时读和写，所以需要将广播接收端放入另一个任务。但每个连接只用一个任务会导致阻塞。更好的方式是：每个连接有两个任务，一个负责读，一个负责写。或者使用 tokio::select! 来同时等待。
我们采用 tokio::select! 来实现非阻塞处理：
use tokio::sync::broadcast;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
async fn handle_connection(
socket: TcpStream,
tx: broadcast::Sender<String>,
) -> Result<(), Box<dyn std::error::Error>> {
let (reader, mut writer) = socket.split();
let mut reader = BufReader::new(reader);
let mut rx = tx.subscribe();
let mut line = String::new();
loop {
tokio::select! {
// 从客户端读取一行
result = reader.read_line(&mut line) => {
match result {
Ok(0) => break, // EOF
Ok(_) => {
let msg = line.trim().to_string();
// 广播消息（包括自己？通常不包括，但这里为了简单，广播给所有，包括自己）
tx.send(msg).unwrap();
line.clear();
}
Err(e) => {
eprintln!("read error: {}", e);
break;
}
}
}
// 从广播通道接收消息
result = rx.recv() => {
match result {
Ok(msg) => {
// 发送给客户端，加上换行符
writer.write_all(format!("{}\n", msg).as_bytes()).await?;
writer.flush().await?;
}
Err(_) => break, // 通道关闭
}
}
}
}
Ok(())
}
第三步：广播消息给所有客户端
我们已经使用了 broadcast::channel，它允许多个发送者和接收者。每个客户端连接持有一个发送者（tx）和一个订阅者（rx）。当某个客户端发送消息时，我们通过 tx.send(msg) 广播，所有订阅者都会收到。
注意：broadcast 的接收者如果落后太多会丢失消息，我们可以通过设置合适的容量或使用其他模式。
完整代码
将上述代码整合到 main.rs，并添加必要的错误处理：
use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::broadcast;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:8080").await?;
println!("Server running on 127.0.0.1:8080");
let (tx, _) = broadcast::channel(10);
loop {
let (socket, addr) = listener.accept().await?;
println!("New client: {}", addr);
let tx = tx.clone();
tokio::spawn(async move {
if let Err(e) = handle_connection(socket, tx).await {
eprintln!("Connection error: {}", e);
}
});
}
}
async fn handle_connection(
socket: TcpStream,
tx: broadcast::Sender<String>,
) -> Result<(), Box<dyn std::error::Error>> {
let (reader, mut writer) = socket.split();
let mut reader = BufReader::new(reader);
let mut rx = tx.subscribe();
let mut line = String::new();
loop {
tokio::select! {
result = reader.read_line(&mut line) => {
match result {
Ok(0) => break,
Ok(_) => {
let msg = line.trim().to_string();
tx.send(msg)?;
line.clear();
}
Err(e) => {
eprintln!("read error: {}", e);
break;
}
}
}
result = rx.recv() => {
match result {
Ok(msg) => {
writer.write_all(format!("{}\n", msg).as_bytes()).await?;
writer.flush().await?;
}
Err(_) => break,
}
}
}
}
Ok(())
}
测试服务器
运行服务器：
cargo run
在另一个终端，使用 nc 或 telnet 连接：
nc 127.0.0.1 8080
打开多个终端，输入消息，观察广播效果。
深入理解：Tokio 任务与调度
Tokio 的任务是轻量级的，每个 tokio::spawn 都会创建一个并发执行的任务。Tokio 使用工作窃取调度器，在多线程运行时中，任务会被分配到不同的线程，实现并行。
重要概念：
- Future：异步计算的结果，惰性的，需要被 poll。
- Executor：驱动 Future 执行的组件。
- Task：一个被调度执行的 Future。
最佳实践与常见陷阱
1. 避免在异步代码中阻塞
不要使用 std::thread::sleep，而是使用 tokio::time::sleep。同样，不要使用阻塞的 Mutex（如 std::sync::Mutex）跨 await 持有，会阻塞整个线程。
推荐：使用 tokio::sync::Mutex 或 RwLock。
2. 小心使用 spawn 的 'static 限制
tokio::spawn 要求任务中的 Future 是 'static 的，意味着不能借用非 'static 数据。通常需要 Arc 共享所有权。
3. 通道容量选择
broadcast 通道容量过小，慢消费者会丢失消息；过大则浪费内存。根据场景调整。
4. 优雅关闭
在服务器中，我们可能需要在收到信号时关闭监听。可以使用 tokio::signal 和 CancellationToken 实现。
use tokio_util::sync::CancellationToken;
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let listener = TcpListener::bind("127.0.0.1:8080").await?;
let token = CancellationToken::new();
let server = tokio::spawn(async move {
loop {
tokio::select! {
_ = token.cancelled() => break,
result = listener.accept() => {
match result {
Ok((socket, addr)) => { /* ... */ }
Err(e) => eprintln!("accept error: {}", e),
}
}
}
}
});
// 等待 Ctrl+C
tokio::signal::ctrl_c().await?;
token.cancel();
server.await??;
Ok(())
}
扩展：添加用户名和私聊
为了展示更多 Tokio 特性，我们可以为聊天室添加用户名。每个客户端连接时发送第一行作为用户名，然后广播通知其他用户。
修改 handle_connection：
async fn handle_connection(
socket: TcpStream,
tx: broadcast::Sender<String>,
) -> Result<(), Box<dyn std::error::Error>> {
// 读取第一行作为用户名
let (reader, mut writer) = socket.split();
let mut reader = BufReader::new(reader);
let mut line = String::new();
reader.read_line(&mut line).await?;
let name = line.trim().to_string();
if name.is_empty() {
return Ok(());
}
println!("{} joined", name);
// ... 其余逻辑类似，但发送消息时加上用户名前缀
}
总结
通过本文，我们学习了如何使用 Tokio 构建一个简单的 TCP 聊天服务器，涵盖了异步 I/O、任务并发、通道通信等核心概念。Tokio 提供了强大的工具集，但正确使用需要深入理解异步模型。建议继续阅读：
- Tokio 官方文档
- 异步流（Async Streams）
- 使用
tokio::select!处理复杂并发
现在，你可以尝试扩展这个项目，例如添加心跳检测、消息持久化或 WebSocket 支持。
延伸阅读
- 若需要处理更复杂的协议，可以考虑使用
tokio-util的Framed。 - 对于 HTTP 服务，可以结合
hyper或axum。
希望这篇文章能帮助你开启 Rust 异步编程之旅。如果你有任何问题，欢迎在评论区讨论。