1 单线程Web服务器

1.1 启动监听、接受连接

  1. use std::net::TcpListener;
  2. pub fn step1(){
  3.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  4.     for stream in listener.incoming(){
  5.         let stream = stream.unwrap();
  6.         println!("接受一个连接");
  7.     }
  8. }

1.2 读取HTTP请求

  1. use std::io::prelude::*;
  2. use std::net::TcpListener;
  3. use std::net::TcpStream;
  5. pub fn step1(){
  6.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  7.     for stream in listener.incoming(){
  8.         handle_connection(stream.unwrap());
  9.     }
  10. }
  12. fn handle_connection(mut stream: TcpStream){
  13.     let mut buf = [0;512];
  14.     let size = stream.read(&mut buf).unwrap();
  15.     println!("读取到HTTP请求: {}字节\n{}\n",size,String::from_utf8_lossy(&buf[..size]));
  16. }

1.3 发送回应

  1. fn handle_connection(mut stream: TcpStream){
  2.     let mut buf = [0;512];
  3.     let size = stream.read(&mut buf).unwrap();
  4.     println!("读取到HTTP请求: {}字节\n{}\n",size,String::from_utf8_lossy(&buf[..size]));
  6.     let ack = "HTTP/1.1 200 OK\r\n";
  7.     stream.write(ack.as_bytes()).unwrap();
  8.     stream.flush().unwrap();
  9. }

1.4 从文件中读取HTML

  1. fn handle_connection(mut stream : TcpStream){
  2.     let mut buf = [0;512];
  3.     let size = stream.read(&mut buf).unwrap();
  4.     println!("读取到HTTP请求: {}字节\n{}\n",size,String::from_utf8_lossy(&buf[..size]));
  6.     let mut file = File::open("./src/index.html").unwrap();
  7.     let mut cont = String::new();
  8.     file.read_to_string(&mut cont).unwrap();
  10.     let mut file = File::open("./src/index.html").unwrap();
  11.     let mut cont = String::new();
  12.     file.read_to_string(&mut cont).unwrap();
  14.     let ack = format!("HTTP/1.1 200 OK\r\n{}",cont);
  15.     stream.write(ack.as_bytes()).unwrap();
  16.     stream.flush().unwrap();
  17. }

1.5 区分请求路径

  1. fn handle_connection(mut stream : TcpStream){
  3.     let mut buf = [0;512];
  5.     let size = stream.read(&mut buf).unwrap();
  6.     println!("读取到HTTP请求: {}字节\n{}\n",size,String::from_utf8_lossy(&buf[..size]));
  8.     let root = b"GET / HTTP/1.1";
  9.     if buf.starts_with(root) {
  10.         let mut file = File::open("./src/index.html").unwrap();
  11.         let mut cont = String::new();
  13.         file.read_to_string(&mut cont).unwrap();
  15.         let ack = format!("HTTP/1.1 200 OK\r\n{}", cont);
  16.         stream.write(ack.as_bytes()).unwrap();
  17.         stream.flush().unwrap();
  18.     }else{
  19.         let mut file = File::open("./src/404.html").unwrap();
  20.         let mut cont = String::new();
  22.         file.read_to_string(&mut cont).unwrap();
  24.         let ack = format!("HTTP/1.1 400 NOT FOUND\r\n{}", cont);
  25.         stream.write(ack.as_bytes()).unwrap();
  26.         stream.flush().unwrap();
  27.     }
  28. }

1.6 提取函数

  1. fn handle_connection(mut stream : TcpStream){
  3.     let mut buf = [0;512];
  5.     let size = stream.read(&mut buf).unwrap();
  6.     println!("读取到HTTP请求: {}字节\n{}\n",size,String::from_utf8_lossy(&buf[..size]));
  8.     let root = b"GET / HTTP/1.1";
  10.     if buf.starts_with(root) {
  11.         return_page(&mut stream,"HTTP/1.1 200 OK","./src/index.html");
  12.     }else{
  13.         return_page(&mut stream,"HTTP/1.1 400 NOT FOUND","./src/404.html");
  14.     }
  15. }
  17. fn return_page(stream : &mut TcpStream,status:&str,file:&str){
  19.     let mut file = File::open(file).unwrap();
  20.     let mut cont = String::new();
  22.     file.read_to_string(&mut cont).unwrap();
  24.     let ack = format!("{}\r\n{}",status, cont);
  25.     stream.write(ack.as_bytes()).unwrap();
  26.     stream.flush().unwrap();
  27. }

2 扩展成多线程

2.1 线程池

  1. use std::thread;
  3. pub struct ThreadPool{
  4.     size : usize,
  5.     workers : Vec<Worker>,
  6. }
  8. struct Worker{
  9.     id: usize,
  10.     handle: thread::JoinHandle<()>,
  11. }
  13. impl Worker{
  14.     pub fn new(id: usize) -> Worker{
  15.         let thread = thread::spawn(|| {});//什么也不做的空线程
  16.         Worker{
  17.             id:id,
  18.             handle:thread,
  19.         }
  20.     }
  21. }
  23. impl ThreadPool{
  25.     pub fn new(max : usize) -> ThreadPool{
  26.         assert!(max > 0);
  28.         // 创建多个工作者
  29.         let mut list = Vec::with_capacity(max);
  30.         for id in 0..max{
  31.             list.push(Worker::new(id));
  32.         }
  34.         ThreadPool{
  35.             size: max,
  36.             workers: list,
  37.         }
  38.     }
  40.     // 后续需要将f传递到Worker中
  41.     pub fn execute<F>(&self, f: F) where F: FnOnce() + Send + 'static {
  42.     }
  43. }
  • std::thread::spawn()的签名如下：
  1. pub fn spawn<F,T>(f: F) -> JoinHandle<T> where F:FnOnce() -> T + Send + 'static{},T: Send + 'static
  • 分析如下
    • 不关心返回值的类型T
    • 参数类型F的特性限定FnOnce()在execute()中也是需要的：因为参数需要最终传递到execute()中
    • 参数类型F的特性限定Send是必须的：Send允许将参数从一个线程传递到另一个线程（新创建的线程）
    • 参数类型F的生命周期限定'static是必须的：不知道线程会运行多长时间

2.2 使用通道传递工作函数

  1. use std::sync;
  2. use std::sync::mpsc;
  3. use std::thread;
  5. // 要传递的工作函数由特性限定表示；但是不能直接传递大小不确定的特性类型，要用Box封装
  6. type Job = Box<FnOnce() + Send + 'static>;
  8. pub struct ThreadPool {
  9.     size: usize,
  10.     workers: Vec<Worker>,
  11.     sender: mpsc::Sender<Job>,
  12. }
  14. struct Worker {
  15.     id: usize,
  16.     handle: thread::JoinHandle<()>,
  17. }
  19. impl Worker {
  20.     pub fn new(id: usize, receiver: sync::Arc<sync::Mutex<mpsc::Receiver<Job>>>) -> Worker {
  21.         let thread = thread::spawn(move || loop {// 需要将receiver的所有权移动到闭包中，所以使用move
  22.             let job = receiver.lock().unwrap().recv().unwrap();
  23.             // 这里无法通过编译: 被封装的 FnOnce() 特性在作为函数进行调用时，需要获取资源所有权
  24.             // 但是Box不允许移出资源的所有权：资源是属于Box的，不能移出
  25.             // 解决方案见下一节
  26.             (*job)();
  27.         });
  28.         Worker {
  29.             id: id,
  30.             handle: thread,
  31.         }
  32.     }
  33. }
  35. impl ThreadPool {
  36.     pub fn new(max: usize) -> ThreadPool {
  37.         assert!(max > 0);
  39.         let (tx, rx) = mpsc::channel();
  40.         let receiver = sync::Arc::new(sync::Mutex::new(rx));
  42.         let mut list = Vec::with_capacity(max);
  44.         for id in 0..max {
  45.             // 不能直接使用receiver，因为所有权被转移走，后续循环中就不能再使用了
  46.             // 使用Arc和Mutex进行封装
  47.             // Arc适用于资源需要被多线程、多处只读使用、无法确定哪部分最后结束使用（然后释放资源）的情况
  48.             // Mutex保证资源的互斥使用
  49.             list.push(Worker::new(id, receiver.clone()));
  50.         }
  52.         ThreadPool {
  53.             size: max,
  54.             workers: list,
  55.             sender: tx,
  56.         }
  57.     }
  59.     pub fn execute<F>(&self, f: F)
  60.     where
  61.         F: FnOnce() + Send + 'static,
  62.     {
  63.         // 用Box封装工作函数，然后发送到通道中
  64.         let job = Box::new(f);
  65.         self.sender.send(job);
  66.     }
  67. }

2.3 使用Box<Self>

  1. use std::sync;
  2. use std::sync::mpsc;
  3. use std::thread;
  5. pub trait FnBox {
  6.     fn call_box(self: Box<Self>);
  7. }
  9. // 总括实现: 用Box<Self>可以取得Box中的值的所有权
  10. impl<F: FnOnce()> FnBox for F {
  11.     fn call_box(self: Box<Self>) {
  12.         (*self)();
  13.     }
  14. }
  16. type Job = Box<FnBox + Send + 'static>;
  18. pub struct ThreadPool {
  19.     sender: mpsc::Sender<Job>,
  20. }
  22. struct Worker;
  24. impl Worker {
  25.     fn run(id: usize, receiver: sync::Arc<sync::Mutex<mpsc::Receiver<Job>>>) {
  26.         thread::spawn(move ||
  27.             loop {
  28.                 // lock()的返回值为LockResult<MutexGuard<T>>类型,其中的MutexGuard<T>失效时，锁被自动释放
  29.                 // let 语句中,对MutexGuard<T>调用recv().unwrap()方法后，将结果赋值给job后，临时的MutexGuard<T>就失效了，锁被释放
  30.                 let job = receiver.lock().unwrap().recv().unwrap();
  31.                 println!("线程{}执行任务", id);
  32.                 job.call_box();//通过Box<Self>类型获取Box中的值的所有权,然后将值作为函数进行调用
  33.             }
  34.             /*
  35.             // while 表达式中的值在整个while块中一直有效，MutexGuard<T>一直有效，锁一直被持有：在调用job.call_box()期间锁也被持有
  36.             // 无法实现多个请求的并行执行
  37.             while let Ok(job) = receiver.lock().unwrap().recv() {
  38.                 println!("Worker {} got a job; executing.", id);
  39.                 job.call_box();
  40.             }
  41.             */
  42.         );
  43.     }
  44. }
  46. impl ThreadPool {
  47.     pub fn new(max: usize) -> ThreadPool {
  48.         assert!(max > 0);
  50.         let (tx, rx) = mpsc::channel();
  51.         let receiver = sync::Arc::new(sync::Mutex::new(rx));
  53.         for id in 0..max {
  54.             Worker::run(id, receiver.clone());
  55.         }
  57.         ThreadPool { sender: tx }
  58.     }
  60.     pub fn execute<F>(&self, f: F)
  61.     where
  62.         F: FnBox + Send + 'static // 这里FnOnce() 改成 FnBox
  63.     {
  64.         let job = Box::new(f);
  65.         self.sender.send(job).unwrap();
  66.     }
  67. }
  • 将第一节的单线程服务器改成多线程时，仅需要修改1.1节的step1()函数
  1. pub fn step1(){
  2.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
  3.     // 创建线程池
  4.     let pool = super::section2::ThreadPool::new(4);
  5.     for stream in listener.incoming(){
  6.         // 使用线程池执行函数
  7.         pool.execute(||handle_connection(stream.unwrap()));
  8.     }
  9. }

3 优雅停机与清理

  1. use std::sync;
  2. use std::sync::mpsc;
  3. use std::thread;
  5. pub trait FnBox {
  6.     fn call_box(self: Box<Self>);
  7. }
  9. impl<F: FnOnce()> FnBox for F {
  10.     // 使用Box<Self>可以从Box<T>中取出T的值，获得所有权
  11.     fn call_box(self: Box<Self>) {
  12.         (*self)();
  13.     }
  14. }
  16. type Job = Box<FnBox + Send + 'static>;
  18. // 使用枚举类型粘结异构类型
  19. enum Msg {
  20.     NewJob(Job),
  21.     Exit,
  22. }
  24. pub struct ThreadPool {
  25.     sender: mpsc::Sender<Msg>,
  26.     list: Vec<Worker>,
  27. }
  29. struct Worker {
  30.     thread: Option<thread::JoinHandle<()>>,
  31. }
  33. impl Worker {
  34.     fn run(id: usize, receiver: sync::Arc<sync::Mutex<mpsc::Receiver<Msg>>>) -> Worker {
  35.         let t = thread::spawn(move || loop {
  36.             // 使用if let
  37.             if let Msg::NewJob(job) = receiver.lock().unwrap().recv().unwrap() {
  38.                 println!("线程{}执行任务", id);
  39.                 job.call_box();
  40.             } else {
  41.                 println!("线程{}收到退出请求", id);
  42.                 break;
  43.             }
  45.             // 使用match
  46.             /*
  47.             let msg = receiver.lock().unwrap().recv().unwrap();
  48.             match msg {
  49.                 Msg::NewJob(job) => {
  50.                     println!("线程{}执行任务", id);
  51.                     job.call_box();
  52.                 }
  53.                 Msg::Exit => {
  54.                     println!("线程{}收到退出请求", id);
  55.                     break;
  56.                 }
  57.             }
  58.             */        });
  60.         Worker { thread: Some(t) }
  61.     }
  62. }
  64. impl ThreadPool {
  65.     pub fn new(max: usize) -> ThreadPool {
  66.         assert!(max > 0);
  68.         let (tx, rx) = mpsc::channel();
  69.         // 使用Arc和Mutex封装类型,使得类型可以安全地在多线程中使用
  70.         let receiver = sync::Arc::new(sync::Mutex::new(rx));
  71.         let mut list = Vec::with_capacity(max);
  73.         for id in 0..max {
  74.             list.push(Worker::run(id, receiver.clone()));
  75.         }
  77.         ThreadPool {
  78.             sender: tx,
  79.             list: list,
  80.         }
  81.     }
  83.     pub fn execute<F>(&self, f: F)
  84.     where
  85.         F: FnBox + Send + 'static,
  86.     {
  87.         let job = Box::new(f);
  88.         self.sender.send(Msg::NewJob(job)).unwrap();
  89.     }
  90. }
  92. impl Drop for ThreadPool {
  93.     fn drop(&mut self) {
  94.         // 发送多个退出请求,让每个线程收到一个
  95.         for _ in &self.list {
  96.             self.sender.send(Msg::Exit).unwrap();
  97.         }
  98.         // 等待每个线程退出
  99.         for item in &mut self.list {
  100.             if let Some(t) = item.thread.take() {
  101.                 match t.join() {
  102.                     Ok(_) => 0,
  103.                     Err(_) => 1,
  104.                 };
  105.             }
  106.         }
  107.     }
  108. }