从单线程到多线程：Rust Web 服务器的性能提升之路

一、问题描述：单线程服务器的局限性

在单线程实现中，服务器对每个请求依次处理。如果有请求需要较长时间才能响应（例如处理复杂的计算或等待外部资源），后续的请求就会被迫等待。为了解释这个问题，我们可以模拟一个“慢请求”。

模拟慢请求

假设我们为请求 /sleep 设计了一个处理分支，服务器在接收到该请求后会延时 5 秒再返回响应。代码示例如下（Listing 21-10）：

match request_line.as_str() {
"GET / HTTP/1.1" => { /* 正常处理 */ },
"GET /sleep HTTP/1.1" => {
// 模拟一个长时间的请求
thread::sleep(Duration::from_secs(5));
// 返回正常页面
},
_ => { /* 返回 404 */ },
}

当同时有请求访问 /sleep 与其他快速响应的页面时，由于服务器是单线程运行，所有请求都将等待慢请求完成后才能处理。这在实际场景中会极大影响用户体验。

二、引入线程池：提升服务器并发处理能力

为了解决单线程服务器的瓶颈，我们需要让服务器能同时处理多个请求。一个常见的解决方案是使用线程池。线程池是一组预先创建好的工作线程，它们在等待任务到来，一旦有新任务，就能从线程池中选取一个空闲线程去执行任务。这样，即使某个请求需要长时间处理，其他线程依然可以处理其他请求。

为什么使用线程池？

• 资源限制：如果每个请求都创建一个新线程，极端情况下可能会导致系统资源耗尽。线程池可以通过预设线程数量来限制并发数量，从而防止拒绝服务（DoS）攻击。
• 提高吞吐量：固定数量的线程能够同时处理多个请求，缓解长时间处理请求带来的阻塞问题。

三、设计线程池接口

在设计线程池时，我们希望它的使用方式与 thread::spawn 类似，这样对调用者来说不需要太多改动。我们的目标接口如下（Listing 21-12）：

let pool = ThreadPool::new(4);
for stream in listener.incoming() {
pool.execute(|| {
handle_connection(stream);
});
}

这个接口简单直观：

• 使用 ThreadPool::new 创建一个指定数量线程的线程池。
• 使用 pool.execute 将需要处理的请求（以闭包形式）提交给线程池执行。

四、基于编译器驱动开发实现线程池

Rust 编译器的严格类型检查能够帮助我们一步步完善设计。我们从创建一个空的 ThreadPool 结构体开始，再根据编译器反馈逐步添加必要的字段和方法。

1. 定义 ThreadPool 结构体

首先我们定义一个最简单的 ThreadPool，这里我们先将项目由二进制 crate 转换为库 crate，将线程池实现放在 src/lib.rs 中（Listing 21-14 前期示例）：

pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}

2. 实现 new 方法

我们需要确保传入的线程数大于 0，否则程序应该 panic（见 Listing 21-13）：

impl ThreadPool {
/// 创建一个新的线程池。
///
/// # Panics
///
/// 如果传入的 `size` 为 0，线程池会 panic。
pub fn new(size: usize) -> ThreadPool {
assert!(size > 0);

// 创建一个通道，用于在线程间传递任务
let (sender, receiver) = mpsc::channel();

// 使用 Arc 和 Mutex 共享 receiver 给每个 Worker
let receiver = Arc::new(Mutex::new(receiver));

let mut workers = Vec::with_capacity(size);

// 根据 size 创建相应数量的 Worker
for id in 0..size {
workers.push(Worker::new(id, Arc::clone(&receiver)));
}

ThreadPool { workers, sender }
}
}

3. 定义 execute 方法

execute 方法接收一个实现 FnOnce() + Send + 'static 的闭包，将其封装为 Job 后发送给线程池（见 Listing 21-19）：

type Job = Box<dyn FnOnce() + Send + 'static>;

impl ThreadPool {
pub fn execute<F>(&self, f: F)
where
F: FnOnce() + Send + 'static,
{
let job = Box::new(f);
self.sender.send(job).unwrap();
}
}

4. 构建 Worker 结构体

Worker 结构体封装了实际的线程，每个 Worker 在创建时都会启动一个线程，并持续等待接收任务执行。Worker 的实现（Listing 21-15 和 Listing 21-20）如下：

use std::thread;
use std::sync::{Arc, Mutex};
use std::sync::mpsc;

struct Worker {
id: usize,
thread: thread::JoinHandle<()>,
}

impl Worker {
fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
let thread = thread::spawn(move || loop {
// 等待从 channel 中接收任务
let job = receiver.lock().unwrap().recv().unwrap();
println!("Worker {} got a job; executing.", id);
job();
});
Worker { id, thread }
}
}

在每个线程中，我们使用 receiver.lock().unwrap().recv().unwrap() 同步地等待任务到来。使用 Arc<Mutex<_>> 保证多个 Worker 能安全共享同一个接收端。

五、整合多线程服务器

将线程池整合到服务器代码中后，服务器的 main 函数会变得类似下面这样：

fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
let pool = ThreadPool::new(4); // 固定 4 个工作线程

for stream in listener.incoming() {
let stream = stream.unwrap();
pool.execute(|| {
handle_connection(stream);
});
}
}

这样一来，无论是快速响应请求（例如 /）还是模拟长时间响应的请求（例如 /sleep），服务器都能同时处理多个请求，互不阻塞。

六、总结与展望

本文详细介绍了如何将单线程 Web 服务器升级为多线程服务器，核心思想是引入线程池来限制并发线程数量，同时利用 Rust 强类型系统和编译器反馈一步步完善实现。主要内容包括：

• 问题分析：单线程服务器在遇到耗时操作时会阻塞后续请求。
• 模拟慢请求：通过 /sleep 请求模拟长时间处理场景，验证单线程问题。
• 线程池设计：设计出类似 ThreadPool::new 和 pool.execute 的 API，使接口与 thread::spawn 保持一致。
• 实现细节：利用 Rust 的 mpsc 通道、Arc 和 Mutex 来在线程间安全传递任务；设计 Worker 结构体来管理每个线程的生命周期。