Rust 学习笔记 - 实战多线程 Web 服务器
构建多线程 Web 服务器
实现流程如下：
在
socket上监听 TCP 连接解析少量的 HTTP 请求
创建一个合适的 HTTP 响应
使用线程池改进服务器的吞吐量
优雅的停机和清理
注意：这并不是最佳实践，而是为了复习前面学习的知识，并了解通用的技术和背后的思路。
实现步骤
Step1 创建项目
创建一个二进制项目
cargo new webserver 复制代码
Step2 监听 TCP 连接
main.rs
fn main() { // 创建监听，出现错误就使用 unwrap 处理，终止执行，如果正常就可以得到一个 TcpListener 的实例 let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); // listener 的 incoming 方法会产生一个返回流序列（TCP Stream）的迭代器 // 有流就代表客户端和服务器建立了一个连接 // for 循环会处理每一个连接并生成一些列的流让我们来处理 for stream in listener.incoming() { let stream = stream.unwrap(); println!("Connection established!"); } } 复制代码运行项目，并在浏览器输入 http://127.0.0.1:7878 访问，现象如下图所示：
Step3 读取请求
main.rs
use std::{net::{TcpListener, TcpStream}, io::Read}; fn main() { // 创建监听，出现错误就使用 unwrap 处理，终止执行，如果正常就可以得到一个 TcpListener 的实例 let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); // listener 的 incoming 方法会产生一个返回流序列（TCP Stream）的迭代器 // 有流就代表客户端和服务器建立了一个连接 // for 循环会处理每一个连接并生成一些列的流让我们来处理 for stream in listener.incoming() { let stream = stream.unwrap(); // 使用函数处理流信息 handle_connection(stream); } } // TCP 内部的状态可能会改变，所以需要标记为 mut fn handle_connection(mut stream: TcpStream) { let mut buffer = [0; 512]; // [0, 0, ...] 512 个字节 stream.read(&mut buffer).unwrap(); // 从流读数据并放入 buffer 缓存区 println!("Request: {}", String::from_utf8_lossy(&buffer)); } 复制代码再次运行项目，并在浏览器输入 http://127.0.0.1:7878 访问，现象如下图所示：
Step4 写响应
main.rs
use std::{net::{TcpListener, TcpStream}, io::{Read, Write}}; // 省略未改动部分 // TCP 内部的状态可能会改变，所以需要标记为 mut fn handle_connection(mut stream: TcpStream) { let mut buffer = [0; 1024]; // [0, 0, ...] 1024 个字节，注意这里设置小了可能会遇到问题 stream.read(&mut buffer).unwrap(); // 从流读数据并放入 buffer 缓存区 let response = "HTTP/1.1 200 OK\r\n\r\n"; stream.write(response.as_bytes()).unwrap(); // flush 会等待并阻止程序运行，直到所有的字节都被写入到连接中 stream.flush().unwrap(); } 复制代码再次运行项目，打开浏览器控制台并输入 http://127.0.0.1:7878 访问，现象如下图所示：
Step5 返回 HTML 文件
在根目录下创建一个 hello.html，内容如下：
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>Hello</title> </head> <body> <h1>Hello!</h1> <p>Hi from Rust</p> </body> </html> 复制代码
main.rs
use std::{net::{TcpListener, TcpStream}, io::{Read, Write}, fs}; // 省略未改动部分 // TCP 内部的状态可能会改变，所以需要标记为 mut fn handle_connection(mut stream: TcpStream) { let mut buffer = [0; 1024]; // [0, 0, ...] 1024 个字节，注意这里设置小了可能会遇到问题 stream.read(&mut buffer).unwrap(); // 从流读数据并放入 buffer 缓存区 let contents = fs::read_to_string("hello.html").unwrap(); let response = format!("HTTP/1.1 200 OK\r\n\r\n{}", contents); stream.write(response.as_bytes()).unwrap(); // flush 会等待并阻止程序运行，直到所有的字节都被写入到连接中 stream.flush().unwrap(); } 复制代码重新运行项目，并在浏览器输入 http://127.0.0.1:7878 访问，现象如下图所示：
Step6 添加对不同路由的简单支持
在根目录下创建一个 404.html，内容如下：
<!DOCTYPE html> <html lang="en"> <head> <meta charset="UTF-8"> <meta http-equiv="X-UA-Compatible" content="IE=edge"> <meta name="viewport" content="width=device-width, initial-scale=1.0"> <title>404</title> </head> <body> <h>Oops!</h> <p>Sorry, 404!</p> </body> </html> 复制代码
main.rs
// 省略未改动部分 // TCP 内部的状态可能会改变，所以需要标记为 mut fn handle_connection(mut stream: TcpStream) { let mut buffer = [0; 1024]; // [0, 0, ...] 1024 个字节，注意这里设置小了可能会遇到问题 stream.read(&mut buffer).unwrap(); // 从流读数据并放入 buffer 缓存区 let get = b"GET / HTTP/1.1\r\n"; // u8 类型的字符串 let (status_line, filename) = if buffer.starts_with(get) { ("HTTP/1.1 200 OK\r\n\r\n", "hello.html") } else { ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html") }; let contents = fs::read_to_string(filename).unwrap(); let response = format!("{}{}", status_line, contents); stream.write(response.as_bytes()).unwrap(); // flush 会等待并阻止程序运行，直到所有的字节都被写入到连接中 stream.flush().unwrap(); } 复制代码重新运行项目，并在浏览器输入 http://127.0.0.1:7878/abc 访问，现象如下图所示：
到此为止，我们已经搭建好一个简单的 Web 服务器了。
Step7 多线程构造
目前我们实现的 Web 服务器是一个单线程的服务器，所有的请求过来之后都需要一个一个处理，如果某一个请求的速度非常慢，就会阻塞后面的请求，这样的服务器性能显然是非常差的，所以我需要进行改善，增加服务器的吞吐量。
Step7.1 模拟慢请求
改善之前我们先来模拟一个比较慢的请求，修改 main.rs 代码如下：
use std::{net::{TcpListener, TcpStream}, io::{Read, Write}, fs, thread, time::Duration}; // 省略未改动部分 // TCP 内部的状态可能会改变，所以需要标记为 mut fn handle_connection(mut stream: TcpStream) { let mut buffer = [0; 1024]; // [0, 0, ...] 1024 个字节，注意这里设置小了可能会遇到问题 stream.read(&mut buffer).unwrap(); // 从流读数据并放入 buffer 缓存区 let get = b"GET / HTTP/1.1\r\n"; // u8 类型的字符串 let sleep = b"GET /sleep HTTP/1.1\r\n"; // /sleep 路由 let (status_line, filename) = if buffer.starts_with(get) { ("HTTP/1.1 200 OK\r\n\r\n", "hello.html") } else if buffer.starts_with(sleep) { thread::sleep(Duration::from_secs(5)); // 休眠 5 秒再返回 ("HTTP/1.1 200 OK\r\n\r\n", "hello.html") } else { ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html") }; let contents = fs::read_to_string(filename).unwrap(); let response = format!("{}{}", status_line, contents); stream.write(response.as_bytes()).unwrap(); // flush 会等待并阻止程序运行，直到所有的字节都被写入到连接中 stream.flush().unwrap(); } 复制代码重新运行项目，并在浏览器输入 http://127.0.0.1:7878/sleep 访问，同时再快速访问 http://127.0.0.1:7878/，这时你就会发现这两个访问都在转圈，直到 5 秒过去，几乎同时刷出来两个访问，目前就是串行的在处理请求。
Step7.2 实现线程池
线程池是一组预分配出来的线程，它们被用于等待，并随时处理可能的任务，当程序接收到一个新的任务的时候会给线程池里的一个线程分配这个任务，这个线程就会处理这个任务，线程池里面其它的线程同时还可以接收其它的任务。线程处理完任务后，我们又会将它放回线程池。所以一个线程池允许你并行的处理连接，从而增加服务器的吞吐量。
main.rs
// 省略未改动部分 use webserver::ThreadPool; fn main() { // 创建监听，出现错误就使用 unwrap 处理，终止执行，如果正常就可以得到一个 TcpListener 的实例 let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); let pool = ThreadPool::new(4); // listener 的 incoming 方法会产生一个返回流序列（TCP Stream）的迭代器 // 有流就代表客户端和服务器建立了一个连接 // for 循环会处理每一个连接并生成一些列的流让我们来处理 for stream in listener.incoming() { let stream = stream.unwrap(); pool.execute(|| { // 使用函数处理流信息 handle_connection(stream); }); } } // 省略未改动部分 复制代码lib.rs
use std::{thread, sync::{mpsc, Arc, Mutex}}; pub struct ThreadPool { workers: Vec<Worker>, sender: mpsc::Sender<Job>, } type Job = Box<dyn FnOnce() + Send + 'static>; impl ThreadPool { /// Create a new ThreadPool /// /// The size is the number of threads in the pool. /// /// # panic /// /// The `new` function will panic if the size is zero. pub fn new(size: usize) -> ThreadPool { assert!(size > 0); let (sender, receiver) = mpsc::channel(); // 使用通道，实现线程内外的数据传递 let receiver = Arc::new(Mutex::new(receiver)); // 为了保证 receiver 能在多个线程中使用和修改 let mut workers = Vec::with_capacity(size); // 创建一个预分配好空间的 vector for id in 0..size { // create some threads and store them in the vector workers.push(Worker::new(id, Arc::clone(&receiver))); } ThreadPool { workers, sender } } pub fn execute<F>(&self, f: F) where // 这里的约束依据 thread::spawn // thread::spawn 的约束为 FnOnce() -> T + Send + 'static // 这里不需要返回什么，所以 FnOnce 就不需要写返回类型了 F: FnOnce() + Send + 'static, { let job = Box::new(f); self.sender.send(job).unwrap(); } } struct Worker { id: usize, // JoinHandle 也是来自于 thread::spawn thread: thread::JoinHandle<()>, } impl Worker { fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker { let thread = thread::spawn(move || loop { let job = receiver.lock().unwrap().recv().unwrap(); println!("Worker {} got a job; executing.", id); job(); }); Worker { id, thread } } } 复制代码执行 cargo doc --open 可以生成文档。
重新运行项目，并在浏览器输入 http://127.0.0.1:7878/sleep 访问，同时再快速访问 http://127.0.0.1:7878/，也可以正常访问，同时可以看到控制台也有输出，每次都使用的是不同的线程。
Step8 优雅的停机和清理
到现在为止，我们可以看一下之前的代码，里面有 3 处警告，说明我们并未直接使用过它们，这也意味着我们还没有清理完所有的东西。
当我们使用 Ctrl + c 使主线程停止的时候，也会使其它线程停止，即便它们正在处理任务，我们要对此进行优化。
优化思路就是为线程池 ThreadPool 实现 drop trait，从而调用每个线程池中线程的 drop 方法，使它们能够在关闭前完成当前正在处理的工作。接着我们还需要通过某种方式来避免线程接收新的请求，并为停机做好准备。
为 ThreadPool 实现 drop trait（lib.rs）：
use std::{thread, sync::{mpsc, Arc, Mutex}}; type Job = Box<dyn FnOnce() + Send + 'static>; enum Message { NewJob(Job), Terminate, } pub struct ThreadPool { workers: Vec<Worker>, sender: mpsc::Sender<Message>, } impl ThreadPool { /// Create a new ThreadPool /// /// The size is the number of threads in the pool. /// /// # panic /// /// The `new` function will panic if the size is zero. pub fn new(size: usize) -> ThreadPool { assert!(size > 0); let (sender, receiver) = mpsc::channel(); // 使用通道，实现线程内外的数据传递 let receiver = Arc::new(Mutex::new(receiver)); // 为了保证 receiver 能在多个线程中使用和修改 let mut workers = Vec::with_capacity(size); // 创建一个预分配好空间的 vector for id in 0..size { // create some threads and store them in the vector workers.push(Worker::new(id, Arc::clone(&receiver))); } ThreadPool { workers, sender } } pub fn execute<F>(&self, f: F) where // 这里的约束依据 thread::spawn // thread::spawn 的约束为 FnOnce() -> T + Send + 'static // 这里不需要返回什么，所以 FnOnce 就不需要写返回类型了 F: FnOnce() + Send + 'static, { let job = Box::new(f); self.sender.send(Message::NewJob(job)).unwrap(); } } impl Drop for ThreadPool { fn drop(&mut self) { for _ in &mut self.workers { self.sender.send(Message::Terminate).unwrap(); } println!("Shutting down all workers."); for worker in &mut self.workers { println!("Shutting down worker {}", worker.id); if let Some(thread) = worker.thread.take() { thread.join().unwrap(); } } } } struct Worker { id: usize, // JoinHandle 也是来自于 thread::spawn thread: Option<thread::JoinHandle<()>>, } impl Worker { fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Message>>>) -> Worker { let thread = thread::spawn(move || loop { let message = receiver.lock().unwrap().recv().unwrap(); match message { Message::NewJob(job) => { println!("Worker {} got a job; executing.", id); job(); } Message::Terminate => { println!("Worker {} was told to terminate.", id); break; } } }); Worker { id, thread: Some(thread) } } } 复制代码看效果可以修改一下 main.rs：
// 省略未改动部分 fn main() { // 创建监听，出现错误就使用 unwrap 处理，终止执行，如果正常就可以得到一个 TcpListener 的实例 let listener = TcpListener::bind("127.0.0.1:7878").unwrap(); let pool = ThreadPool::new(4); // listener 的 incoming 方法会产生一个返回流序列（TCP Stream）的迭代器 // 有流就代表客户端和服务器建立了一个连接 // for 循环会处理每一个连接并生成一些列的流让我们来处理 for stream in listener.incoming().take(2) { // 接收两次请求后自动终止 let stream = stream.unwrap(); pool.execute(|| { // 使用函数处理流信息 handle_connection(stream); }); } println!("Shutting down"); } // 省略未改动部分
作者：Eagle_Clark
链接：https://juejin.cn/post/7042581639204962334
玩站网免费分享SEO网站优化 技术及文章 伪原创工具 https://www.237it.com/