基于单线程模拟慢请求

  1. use std::{
  2.     fs,
  3.     io::{prelude::*, BufReader},
  4.     net::{TcpListener, TcpStream},
  5.     thread,
  6.     time::Duration,
  7. };
  9. fn main() {
  10.     // 监听地址: 127.0.0.1:7878
  11.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  13.     // 建立连接
  14.     for stream in listener.incoming() {
  15.         let stream = stream.unwrap();
  17.         handle_connection(stream);
  18.     }
  19. }
  21. fn handle_connection(mut stream: TcpStream) {
  22.     let buf_reader = BufReader::new(&mut stream);
  23.     let request_line = buf_reader.lines().next().unwrap().unwrap();
  25.     // let (status_line, filename) = if request_line == "GET / HTTP/1.1" {
  26.     //     ("HTTP/1.1 200 OK", "hello.html")
  27.     // } else {
  28.     //     ("HTTP/1.1 404 NOT FOUND", "404.html")
  29.     // };
  31.     let (status_line, filename) = match &request_line[..] {
  32.         "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
  33.         "GET /sleep HTTP/1.1" => {
  34.             thread::sleep(Duration::from_secs(5));
  35.             ("HTTP/1.1 200 OK", "hello.html")
  36.         }
  37.         _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
  38.     };
  40.     let contents = fs::read_to_string(filename).unwrap();
  41.     let length = contents.len();
  43.     let response =
  44.         format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
  46.     stream.write_all(response.as_bytes()).unwrap();
  47. }
如果我们连续访问 / 路径，那效果跟之前一样：立刻看到请求的页面。但假如先访问 /sleep ，接着在另一个页面访问 /，就会看到 / 的页面直到 5 秒后才会刷出来，验证了请求排队这个糟糕的事实。 至于如何解决，其实办法不少，本章我们来看看一个经典解决方案：线程池。

使用线程池改善吞吐

为每个请求生成一个线程

  1. fn main() {
  2.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  4.     for stream in listener.incoming() {
  5.         let stream = stream.unwrap();
  7.         thread::spawn(|| {
  8.             handle_connection(stream);
  9.         });
  10.     }
  11. }
这种实现下，依次访问 /sleep 和 / 就无需再等待，不错的开始。这显然不是我们的最终方案，原因在于它会生成无上限的线程数，最终导致资源耗尽。但它确实是一个好的起点。

限制创建线程的数量

我们需要限制线程池中的线程数量，以保护服务器免受拒绝服务攻击（DoS）的影响：如果针对每个请求创建一个新线程，那么一个人向我们的服务器发出1000万个请求，会直接耗尽资源，导致后续用户的请求无法被处理，这也是拒绝服务名称的来源。架构设计首先是设定最大线程数的上限，其次维护一个请求队列。池中的线程去队列中依次弹出请求并处理。这样就可以同时并发处理 N 个请求，其中 N 是线程数。 
  1. use std::{
  2.     fs,
  3.     io::{prelude::*, BufReader},
  4.     net::{TcpListener, TcpStream},
  5.     thread,
  6.     time::Duration,
  7. };
  9. use multi_thread::ThreadPool;
  11. fn main() {
  12.     // 监听地址: 127.0.0.1:7878
  13.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  14.     // 创建一个包含 4 个线程的线程池
  15.     let pool = ThreadPool::new(4);
  17.     // 建立连接
  18.     for stream in listener.incoming() {
  19.         let stream = stream.unwrap();
  21.         // 分发执行请求
  22.         pool.execute(|| {
  23.             handle_connection(stream);
  24.         });
  25.     }
  26. }
  28. fn handle_connection(mut stream: TcpStream) {
  29.     let buf_reader = BufReader::new(&mut stream);
  30.     let request_line = buf_reader.lines().next().unwrap().unwrap();
  32.     // let (status_line, filename) = if request_line == "GET / HTTP/1.1" {
  33.     //     ("HTTP/1.1 200 OK", "hello.html")
  34.     // } else {
  35.     //     ("HTTP/1.1 404 NOT FOUND", "404.html")
  36.     // };
  38.     let (status_line, filename) = match &request_line[..] {
  39.         "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
  40.         "GET /sleep HTTP/1.1" => {
  41.             thread::sleep(Duration::from_secs(5));
  42.             ("HTTP/1.1 200 OK", "hello.html")
  43.         }
  44.         _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
  45.     };
  47.     let contents = fs::read_to_string(filename).unwrap();
  48.     let length = contents.len();
  50.     let response =
  51.         format!("{status_line}\r\nContent-Length: {length}\r\n\r\n{contents}");
  53.     stream.write_all(response.as_bytes()).unwrap();
  54. }
创建 src/lib.rs 文件并写入如下代码: 
  1. use std::{
  2.   sync::{mpsc, Arc, Mutex},
  3.   thread,
  4. };
  6. type Job = Box<dyn FnOnce() + Send + 'static>;
  9. pub struct ThreadPool {
  10.   workers: Vec<Worker>,
  11.   sender: mpsc::Sender<Job>,
  12. }
  14. impl ThreadPool {
  15.   // 初始化实例
  16.   pub fn new(size: usize) -> ThreadPool {
  17.     assert!(size > 0);
  19.     let (sender, receiver) = mpsc::channel();
  21.     let receiver = Arc::new(Mutex::new(receiver));
  23.     let mut workers = Vec::with_capacity(size);
  25.     for id in 0..size {
  26.         workers.push(Worker::new(id, Arc::clone(&receiver)));
  27.     }
  29.     ThreadPool { workers, sender }
  30.   }
  31.   pub fn execute<F>(&self, f: F)
  32.   where
  33.       F: FnOnce() + Send + 'static,
  34.   {
  35.     let job = Box::new(f);
  37.     self.sender.send(job).unwrap();
  38.   }
  39. }
  41. // 创建一个 Worker 结构体，作为 ThreadPool 和任务线程联系的桥梁，它的任务是获得将要执行的代码，然后在具体的线程中去执行。想象一个场景：一个餐馆，Worker 等待顾客的点餐，然后将具体的点餐信息传递给厨房，感觉类似服务员,
  42. // 由于外部调用者无需知道 Worker 的存在，因此这里使用了私有的声明。
  43. struct Worker {
  44.   id: usize,
  45.   thread: thread::JoinHandle<()>,
  46. }
  48. impl Worker {
  49.   fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
  50.     let thread = thread::spawn(move || loop {
  51.       let job = receiver.lock().unwrap().recv().unwrap();
  53.       println!("Worker {id} got a job; executing.");
  55.       job();
  56.     });
  57.     Worker { id, thread }
  58.   }
  59. }