Tokio的使用

2018年08月10日留言

请用通俗易懂的语言，讲一下rust和tokio的关系？为什么需要tokio？官方为什么不提供？

Rust 和 Tokio 的关系可以用一个很形象的比喻来解释：

Rust 就像是一个工厂，它提供了制造各种产品的工具和材料，但它本身不会告诉你具体要生产什么产品。

Tokio 就像是专门的生产线，它告诉这个工厂：”我们要高效地同时处理很多任务，比如同时接收很多网络请求”。

为什么需要 Tokio？

想象你开了一家餐厅：

传统方式（同步编程）：一个服务员只能服务一桌客人，客人点餐后，服务员就站在那里等厨房做菜，什么都不干。如果有100桌客人，你就需要100个服务员。

Tokio方式（异步编程）：一个服务员可以同时服务多桌客人。给A桌点完餐后，不用等菜做好，立刻去服务B桌，然后C桌…当A桌的菜好了，再回去上菜。这样几个服务员就能服务很多桌客人。

在程序里：

网络请求、文件读写、数据库查询就像”等厨房做菜”
如果用传统方式，程序会”傻等”，浪费CPU资源
Tokio让程序可以在等待的时候去做其他事情

为什么官方不提供？

这其实体现了Rust的设计哲学：

保持核心精简：Rust标准库只提供最基础的功能，就像提供基础工具，而不是具体的产品
选择自由：异步编程有很多不同的实现方式，Tokio只是其中最流行的一种。还有其他的如async-std等。如果官方强制选择一种，就限制了开发者的选择
避免臃肿：不是所有程序都需要异步功能。如果你只是写个简单的命令行工具，就不需要Tokio的复杂功能
社区驱动：让社区来开发和竞争，通常能产生更好的解决方案

这就像汽车厂商专注于制造发动机和底盘（Rust标准库），而让专业公司来做导航系统、音响系统（各种第三方库）。每个人都专注于自己最擅长的领域，最终用户可以自由组合需要的功能。

所以 Tokio 并不是 Rust 的一部分，而是一个非常优秀的第三方异步运行时库，它和 Rust 标准库配合得很好，成为了 Rust 异步编程的事实标准。

我来用一个典型的Web服务器例子来说明Tokio的使用和优势：让我用一个生动的场景来解释这个例子：

场景：网上书店的订单处理系统

假设你的网站需要处理用户下单，每次下单都要：

检查库存（查询数据库，耗时2秒）
返回结果给用户

同步方式的问题

就像只有一个收银员的书店：

第1个顾客来买书，收银员去仓库查库存（2秒）
在这2秒内，其他顾客只能排队等待
如果同时来10个顾客，最后一个要等20秒！

资源消耗：

每个请求需要一个线程
10个并发请求 = 10个线程
1000个并发请求 = 1000个线程（系统会崩溃！）

Tokio异步方式的优势

就像聪明的收银员：

第1个顾客来买书，收银员告诉仓库”查A书的库存”，然后立刻服务第2个顾客
第2个顾客来买书，告诉仓库”查B书的库存”，立刻服务第3个顾客
当A书的库存信息回来时，收银员立刻告诉第1个顾客结果
同时来10个顾客，大约2秒后所有人都能得到结果！

实际性能对比：

同步服务器处理10个请求：
请求1: 2秒后完成
请求2: 4秒后完成  
请求3: 6秒后完成
...
请求10: 20秒后完成
总耗时: 20秒

异步服务器处理10个请求：
所有请求几乎同时开始处理
所有请求: 约2秒后全部完成
总耗时: 2秒

实际应用场景

典型的需要Tokio的场景：

Web API服务器：同时处理成百上千个HTTP请求
聊天应用：同时维持大量WebSocket连接
爬虫程序：同时抓取多个网页
数据处理：同时读取多个文件或数据库

不需要Tokio的场景：

简单的命令行工具
CPU密集型计算（没有等待时间）
单线程的小程序

核心优势总结

资源高效：一个线程可以处理成千上万个连接
响应快速：不会因为一个慢请求影响其他请求
扩展性强：轻松处理高并发场景
内存友好：异步任务比线程轻量得多

这就是为什么像Discord、Dropbox等公司选择Rust+Tokio来构建高性能后端服务的原因！

用代码说明差距

1. 创建项目目录

1
2
3

# 创建项目
cargo new tokio_performance_test
cd tokio_performance_test

2. 文件放置位置

tokio_performance_test/
├── Cargo.toml         
├── src/
│   └── main.rs       
└── README.md

3. 替换文件内容

Cargo.toml:

[package]
name = "tokio_performance_test"
version = "0.1.0"
edition = "2024"

[dependencies]
tokio = { version = "1.0", features = ["full"] }
reqwest = { version = "0.11", features = ["json"] }

src/main.rs:

use std::io::prelude::*;
use std::net::{TcpListener, TcpStream};
use std::thread;
use std::time::{Duration, Instant};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener as AsyncTcpListener, TcpStream as AsyncTcpStream};

// ==================== 真正的同步服务器（单线程）====================
fn handle_sync_connection_single_thread(mut stream: TcpStream) {
    let mut buffer = [0; 1024];
    let _ = stream.read(&mut buffer);

    // 模拟数据库查询 - 阻塞1秒
    thread::sleep(Duration::from_secs(1));

    let response = "HTTP/1.1 200 OK\r\n\r\nSync Single Thread Response";
    let _ = stream.write(response.as_bytes());
    let _ = stream.flush();
}

fn start_single_thread_sync_server() {
    thread::spawn(|| {
        let listener = TcpListener::bind("127.0.0.1:8003").unwrap();
        println!("🐌 单线程同步服务器启动在 127.0.0.1:8003");

        for stream in listener.incoming() {
            match stream {
                Ok(stream) => {
                    // 注意：这里不创建新线程，真正的串行处理！
                    handle_sync_connection_single_thread(stream);
                }
                Err(_) => {}
            }
        }
    });
}
fn handle_sync_connection(mut stream: TcpStream) {
    let mut buffer = [0; 1024];
    let _ = stream.read(&mut buffer);

    // 模拟数据库查询 - 阻塞1秒
    thread::sleep(Duration::from_secs(1));

    let response = "HTTP/1.1 200 OK\r\n\r\nSync Server Response";
    let _ = stream.write(response.as_bytes());
    let _ = stream.flush();
}

fn start_sync_server() {
    thread::spawn(|| {
        let listener = TcpListener::bind("127.0.0.1:8001").unwrap();
        println!("🔄 同步服务器启动在 127.0.0.1:8001");

        for stream in listener.incoming() {
            match stream {
                Ok(stream) => {
                    thread::spawn(move || {
                        handle_sync_connection(stream);
                    });
                }
                Err(_) => {}
            }
        }
    });
}

// ==================== 异步服务器 ====================
async fn handle_async_connection(mut stream: AsyncTcpStream) {
    let mut buffer = [0; 1024];
    let _ = stream.read(&mut buffer).await;

    // 模拟数据库查询 - 异步等待1秒
    tokio::time::sleep(Duration::from_secs(1)).await;

    let response = "HTTP/1.1 200 OK\r\n\r\nAsync Server Response";
    let _ = stream.write_all(response.as_bytes()).await;
}

async fn start_async_server() {
    tokio::spawn(async {
        let listener = AsyncTcpListener::bind("127.0.0.1:8002").await.unwrap();
        println!("⚡ 异步服务器启动在 127.0.0.1:8002");

        loop {
            match listener.accept().await {
                Ok((stream, _)) => {
                    tokio::spawn(async move {
                        handle_async_connection(stream).await;
                    });
                }
                Err(_) => {}
            }
        }
    });
}

// ==================== 性能测试 ====================
async fn simple_http_test(addr: &str, num_requests: usize, test_name: &str) {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    use tokio::net::TcpStream;

    println!("\n🧪 {} - {}个请求", test_name, num_requests);
    let start = Instant::now();

    let mut tasks = vec![];

    for i in 0..num_requests {
        let addr = addr.to_string();
        let task = tokio::spawn(async move {
            match TcpStream::connect(&addr).await {
                Ok(mut stream) => {
                    let request = "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n";
                    let _ = stream.write_all(request.as_bytes()).await;

                    let mut response = vec![0; 1024];
                    let _ = stream.read(&mut response).await;

                    println!("  ✅ 请求 {} 完成", i + 1);
                }
                Err(e) => {
                    println!("  ❌ 请求 {} 失败: {}", i + 1, e);
                }
            }
        });
        tasks.push(task);
    }

    for task in tasks {
        let _ = task.await;
    }

    let duration = start.elapsed();
    println!("📊 {} 总耗时: {:.2}秒", test_name, duration.as_secs_f64());
}

#[tokio::main]
async fn main() {
    println!("🚀 启动服务器性能对比测试\n");

    // 启动三个服务器
    start_sync_server(); // 多线程同步服务器
    start_single_thread_sync_server(); // 单线程同步服务器
    start_async_server().await; // 异步服务器

    // 等待服务器启动
    tokio::time::sleep(Duration::from_secs(2)).await;

    println!("\n{}", "=".repeat(50));
    println!("开始性能测试...");

    // 测试参数 - 可以修改这个数字来测试不同的并发数
    let num_requests = 1000; // 改为1000个并发请求！

    // 测试多线程同步服务器
    simple_http_test("127.0.0.1:8001", num_requests, "多线程同步服务器").await;
    tokio::time::sleep(Duration::from_secs(1)).await;

    // 测试单线程同步服务器（真正的串行）
    let small_test = 5; // 只测试5个，不然要等太久
    simple_http_test("127.0.0.1:8003", small_test, "单线程同步服务器").await;
    tokio::time::sleep(Duration::from_secs(1)).await;

    // 测试异步服务器
    simple_http_test("127.0.0.1:8002", num_requests, "异步服务器").await;

    println!("\n{}", "=".repeat(50));
    println!("📋 测试总结:");
    println!("• 每个请求都模拟1秒的数据库查询时间");
    println!(
        "• 多线程同步服务器: 为每个连接创建线程，{}个请求≈1秒",
        num_requests
    );
    println!("• 单线程同步服务器: 真正的串行处理，5个请求≈5秒");
    println!(
        "• 异步服务器: 可以并发处理多个请求，{}个请求≈1秒",
        num_requests
    );
    println!("\n💡 关键区别:");
    println!("• 多线程同步: {}个线程，内存消耗大", num_requests);
    println!("• 异步: 1个线程，内存消耗小，扩展性更好");
    println!("• 当并发数达到10万+时，异步的优势就非常明显了！");
    println!("\n💡 提示: 修改 main() 函数中的 num_requests 变量来测试更多并发请求");
}

README.md:


# Tokio 性能测试项目

这个项目演示了 Rust 同步编程和 Tokio 异步编程在处理并发请求时的性能差异。

## 项目结构


tokio_performance_test/
├── Cargo.toml          # 项目配置和依赖
├── src/
│   └── main.rs         # 主程序代码
└── README.md           # 本说明文档


## 快速开始

### 1. 创建项目

cargo new tokio_performance_test
cd tokio_performance_test


### 2. 替换文件内容
- 将 `Cargo.toml` 的内容替换为提供的配置
- 将 `src/main.rs` 的内容替换为提供的代码

### 3. 运行测试

cargo run


## 执行结果:

```shell
🚀 启动服务器性能对比测试

🔄 同步服务器启动在 127.0.0.1:8001
🐌 单线程同步服务器启动在 127.0.0.1:8003
⚡ 异步服务器启动在 127.0.0.1:8002

==================================================
开始性能测试...

🧪 多线程同步服务器 - 1000个请求
  ✅ 请求 6 完成
  ✅ 请求 8 完成
  ✅ 请求 28 完成
  ✅ 请求 23 完成
  ✅ 请求 43 完成
  ✅ 请求 13 完成
  ✅ 请求 19 完成
  ✅ 请求 14 完成
  ✅ 请求 25 完成
  ✅ 请求 22 完成
  ✅ 请求 21 完成
  ✅ 请求 40 完成
  
  // 为节省篇幅省略部分输出
  ✅ 请求 706 完成
  ✅ 请求 721 完成
  ✅ 请求 701 完成
  ✅ 请求 694 完成
  ✅ 请求 718 完成
  ✅ 请求 717 完成
  ✅ 请求 715 完成
📊 多线程同步服务器 总耗时: 1.12秒

🧪 单线程同步服务器 - 5个请求
  ✅ 请求 2 完成
  ✅ 请求 1 完成
  ✅ 请求 3 完成
  ✅ 请求 4 完成
  ✅ 请求 5 完成
📊 单线程同步服务器 总耗时: 5.02秒

🧪 异步服务器 - 1000个请求
  ✅ 请求 17 完成
  ✅ 请求 8 完成
  ✅ 请求 7 完成
  ✅ 请求 1 完成
  ✅ 请求 6 完成
  ✅ 请求 2 完成
  ✅ 请求 3 完成
  ✅ 请求 4 完成
  ✅ 请求 12 完成
  ✅ 请求 53 完成
  ✅ 请求 109 完成
  ✅ 请求 42 完成
  ✅ 请求 5 完成
  ✅ 请求 30 完成
  
  // 为节省篇幅省略部分输出
  ✅ 请求 161 完成
  ✅ 请求 170 完成
  ✅ 请求 149 完成
  ✅ 请求 164 完成
  ✅ 请求 163 完成
  ✅ 请求 167 完成
  ✅ 请求 177 完成
  ✅ 请求 160 完成
  ✅ 请求 171 完成
  ✅ 请求 178 完成
📊 异步服务器 总耗时: 1.19秒

==================================================
📋 测试总结:
• 每个请求都模拟1秒的数据库查询时间
• 多线程同步服务器: 为每个连接创建线程，1000个请求≈1秒
• 单线程同步服务器: 真正的串行处理，5个请求≈5秒
• 异步服务器: 可以并发处理多个请求，1000个请求≈1秒

💡 关键区别:
• 多线程同步: 1000个线程，内存消耗大
• 异步: 1个线程，内存消耗小，扩展性更好
• 当并发数达到10万+时，异步的优势就非常明显了！

💡 提示: 修改 main() 函数中的 num_requests 变量来测试更多并发请求

实际应用场景

该测试模拟了真实 Web 服务中的场景：

API 服务器处理大量并发请求
每个请求都需要查询数据库或调用外部服务
网络 I/O 操作存在延迟

技术要点

同步编程: 每个请求阻塞一个线程，资源消耗大
异步编程: 单个线程可以处理多个请求，资源利用率高
Tokio: Rust 生态系统中最成熟的异步运行时

依赖说明

tokio: 异步运行时，提供异步 I/O 操作
reqwest: HTTP 客户端库（如果需要发送 HTTP 请求的话）

🧪 多线程同步服务器 - 1000个请求
📊 多线程同步服务器总耗时: 1.05秒

🧪 单线程同步服务器 - 5个请求
📊 单线程同步服务器总耗时: 5.02秒 ← 这个才是真正的串行！

🧪 异步服务器 - 1000个请求
📊 异步服务器总耗时: 1.03秒


## 真正的区别在这里：

1. **资源消耗**：
   - 多线程同步：1000个请求 = 1000个线程 = 大约8GB内存
   - 异步：1000个请求 = 1个线程 = 几MB内存

2. **系统限制**：
   - 线程数有上限（通常几千个），超过后系统崩溃
   - 异步可以轻松处理几十万个并发连接

3. **上下文切换**：
   - 1000个线程频繁切换，CPU开销大
   - 异步无需线程切换，CPU利用率高

原文链接: https://dashen.tech/2018/08/10/Tokio的使用/

版权声明: 转载请注明出处.

清澄秋爽

苹果树下的思索者书写是对思维的缓存

Tokio的使用

请用通俗易懂的语言，讲一下rust和tokio的关系？为什么需要tokio？官方为什么不提供？

为什么需要 Tokio？

为什么官方不提供？

场景：网上书店的订单处理系统

同步方式的问题

Tokio异步方式的优势

实际应用场景

核心优势总结

1. 创建项目目录

2. 文件放置位置

3. 替换文件内容

实际应用场景

技术要点

依赖说明

文章目录