Rust Tokio 入门：异步 I/O 与网络编程基础——手把手写 TCP/UDP 服务器

不吃草的牛德

发布于 2026-04-23 13:00:05

410

文章被收录于专栏：RustRust

Tokio 系列已经更新到第四篇了！

前三篇我们分别讲了：

• 为什么 Rust 需要异步运行时
• Future、Task、Runtime 三大核心概念
• 运行时配置与调优实战

从今天开始，我们正式进入实战环节 —— 用 Tokio 进行异步网络编程。

本篇重点：异步 TCP 和 UDP 服务器入门。我们会一步步写一个可运行的 TCP Echo Server，再顺便实现一个简单的 UDP Echo Server，让你彻底掌握 tokio::net 模块的使用。

所有代码已在 Tokio 1.50+ 上测试通过，直接复制即可运行。

一、为什么网络编程特别适合异步？

网络 I/O 的特点是：

• 大量时间花在“等待”上（等待客户端连接、等待数据到达、等待发送缓冲区可用）
• 数据到达是事件驱动的（epoll/kqueue 通知）

异步模型完美匹配：遇到 I/O 就 .await 让出线程，去服务其他连接，事件就绪后再被唤醒继续处理。

Tokio 把这一切封装得非常优雅，你写出来的代码几乎和同步代码一样清晰，但性能却能达到 C/C++ 的级别。

二、核心模块：tokio::net

Tokio 网络编程主要使用 tokio::net 模块，提供了异步版本的：

• TcpListener / TcpStream
• UdpSocket
• UnixListener / UnixStream（Unix 域套接字）

这些类型都实现了 AsyncRead 和 AsyncWrite trait，支持 .await 风格的读写操作。

三、手把手写一个高并发 TCP Echo Server

需求：客户端发什么，服务器原样返回。支持同时处理成千上万的连接。

步骤 1：创建项目并添加依赖

# Cargo.toml
[dependencies]
tokio = {version="1.50.0", features = ["full"]}

步骤 2：完整代码（推荐直接复制运行）

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // 1. 绑定地址，创建监听器
    let listener = TcpListener::bind("0.0.0.0:8080").await?;
    println!("TCP Echo Server 启动成功，监听端口: 8080");

    loop {
        // 2. 等待新客户端连接（异步，非阻塞）
        let (mut socket, addr) = listener.accept().await?;
        println!("新客户端连接: {}", addr);

        // 3. 为每个连接 spawn 一个独立 Task，实现并发
        tokio::spawn(async move {
            let mut buf = [0; 1024];

            // 持续处理这个连接的数据
            loop {
                // 4. 异步读取数据socket.write_all
                // 注意：TCP 是面向流的，不是面向包的
                let n = socket.read(&mut buf).await.map_err(|e| {
                    eprintln!("读取异常: {}", e);
                    e
                })?;
                if n == 0 { return; } // 客户端正常关闭

                // 5. 把读到的数据原样写回去（Echo）
                if let Err(e) = socket.write_all(&buf[0..n]).await {
                    eprintln!("写入错误: {}", e);
                    return;
                }

                // 注意：直接写 TcpStream 不需要手动 flush，除非使用了 BufWriter
                
            }
        });
    }
}

运行方式：

cargo run

然后用 telnet 或 nc 测试：

nc 127.0.0.1 8080
# 输入任意内容，回车后会原样返回

为什么能高并发？

• listener.accept().await 是异步的，不会阻塞整个服务器。
• 每个客户端连接都用 tokio::spawn 变成独立 Task，由 Runtime 统一调度。
• 少量 worker 线程就能同时服务大量连接。

四、代码改进版：加上优雅错误处理与超时

生产环境中我们通常会加上超时和更健壮的错误处理：

use tokio::time::{timeout, Duration};

async fn handle_client(mut socket: tokio::net::TcpStream, addr: std::net::SocketAddr) {
    let mut buf = vec![0; 4096];  // 使用 Vec 更灵活

    loop {
        // 设置 30 秒读取超时
        match timeout(Duration::from_secs(30), socket.read(&mut buf)).await {
            Ok(Ok(0)) => {
                println!("客户端 {} 正常断开", addr);
                break;
            }
            Ok(Ok(n)) => {
                if let Err(e) = socket.write_all(&buf[0..n]).await {
                    eprintln!("写入失败 {}: {}", addr, e);
                    break;
                }
            }
            Ok(Err(e)) => {
                eprintln!("读取失败 {}: {}", addr, e);
                break;
            }
            Err(_) => {
                println!("客户端 {} 读取超时，断开连接", addr);
                break;
            }
        }
    }
}

// 在主循环中调用：
let (socket, addr) = listener.accept().await?;
tokio::spawn(handle_client(socket, addr));

代码中每次 accept 都分配并初始化（置零） 4KB 内存，在高并发下会有明显的 CPU 开销。可以使用 BytesMut进行优化。高性能网络编程中，基本上大家都在用 BytesMut避免不必要的内存拷贝。

use tokio::net::{TcpListener, TcpStream};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use bytes::{BytesMut, Buf}; // 引入 Buf trait 用于处理字节流
use std::net::SocketAddr;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("0.0.0.0:8080").await?;
    println!("高性能 TCP Echo Server (BytesMut 版) 启动: 8080");

    loop {
        let (socket, addr) = listener.accept().await?;
        // 每一个连接依然分配一个独立 Task
        tokio::spawn(async move {
            if let Err(e) = handle_client(socket, addr).await {
                eprintln!("客户端 {} 处理异常: {}", addr, e);
            }
        });
    }
}

async fn handle_client(mut socket: TcpStream, addr: SocketAddr) -> io::Result<()> {
    // 优化点：使用 BytesMut 预分配容量（4KB），但不立即初始化内存
    let mut buf = BytesMut::with_capacity(4096);

    loop {
        // read_buf 是针对 BytesMut 的特殊优化方法
        // 它会自动扩容，并只在有数据读取时才推进内部指针
        let n = socket.read_buf(&mut buf).await?;

        if n == 0 {
            println!("客户端 {} 正常断开", addr);
            return Ok(());
        }

        // 此时 buf 中包含了读取到的 n 字节数据
        // 直接将这部分数据（切片）写回
        socket.write_all(&buf).await?;

        // 关键：清空缓冲区以备下次读取
        // 注意：clear() 只是移动指针，不会释放或重新初始化底层内存，实现复用
        buf.clear();
    }
}

五、UDP Echo Server（更简单）

UDP 是无连接的，实现起来比 TCP 还简单：

use tokio::net::UdpSocket;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let socket = UdpSocket::bind("0.0.0.0:8081").await?;
    println!("UDP Echo Server 启动，监听端口: 8081");

    let mut buf = [0u8; 1024];

    loop {
        let (len, addr) = socket.recv_from(&mut buf).await?;
        
        // 原样返回
        socket.send_to(&buf[0..len], addr).await?;
        
        println!("收到来自 {} 的 {} 字节数据", addr, len);
    }
}

测试：

# 发送测试
echo -n "Hello Tokio" | nc -u 127.0.0.1 8081

六、异步 I/O 重要 trait 速览

• AsyncRead / AsyncWrite：异步读写核心 trait
• AsyncReadExt / AsyncWriteExt：提供了 read、write_all、read_to_end 等便捷方法（带 .await）
• BufReader / BufWriter：推荐用于减少系统调用（后续文章会详细介绍）