Rust编程语言入门之最后的项目：多线程 Web 服务器

最后的项目：多线程 Web 服务器

构建多线程 Web 服务器

• 在 socket 上监听 TCP 连接
  
• 解析少量的 HTTP 请求
  
• 创建一个合适的 HTTP 响应
  
• 使用线程池改进服务器的吞吐量
  
• 优雅的停机和清理
  
• 注意：并不是最佳实践
  

创建项目

1. ~/rust
2. ➜ cargo new hello
3.      Created binary (application) `hello` package
5. ~/rust
6. ➜

复制代码

main.rs 文件

1. use std::net::TcpListener;
3. fn main() {
4.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
5.     for stream in listener.incoming() {
6.         let stream = stream.unwrap();
8.         println!("Connection established!");
9.     }
10. }

复制代码

修改一：

1. use std::io::prelude::*;
2. use std::net::TcpListener;
3. use std::net::TcpStream;
5. fn main() {
6.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
7.     for stream in listener.incoming() {
8.         let stream = stream.unwrap();
10.         handle_connection(stream);
11.     }
12. }
14. fn handle_connection(mut stream: TcpStream) {
15.     let mut buffer = [0; 512];
17.     stream.read(&mut buffer).unwrap();
19.     println!("Request: {}", String::from_utf8_lossy(&buffer[..]));
20. }

复制代码

修改二：

1. use std::io::prelude::*;
2. use std::net::TcpListener;
3. use std::net::TcpStream;
5. fn main() {
6.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
7.     for stream in listener.incoming() {
8.         let stream = stream.unwrap();
10.         handle_connection(stream);
11.     }
12. }
14. fn handle_connection(mut stream: TcpStream) {
15.     let mut buffer = [0; 512];
17.     stream.read(&mut buffer).unwrap();
19.     // 请求
20.     // Method Request-URI HTTP-Version CRLF
21.     // headers CRLF
22.     // message-body
24.     // 响应
25.     // HTTP-Version Status-Code Reason-Phrase CRLF
26.     // headers CRLF
27.     // message-body
29.     let response = "HTTP/1.1 200 OK\r\n\r\n";
31.     stream.write(response.as_bytes()).unwrap();
32.     stream.flush().unwrap();
34.     println!("Request: {}", String::from_utf8_lossy(&buffer[..]));
35. }

复制代码

修改三：

1. use std::fs;
2. use std::io::prelude::*;
3. use std::net::TcpListener;
4. use std::net::TcpStream;
6. fn main() {
7.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
8.     for stream in listener.incoming() {
9.         let stream = stream.unwrap();
11.         handle_connection(stream);
12.     }
13. }
15. fn handle_connection(mut stream: TcpStream) {
16.     let mut buffer = [0; 512];
18.     stream.read(&mut buffer).unwrap();
20.     // 请求
21.     // Method Request-URI HTTP-Version CRLF
22.     // headers CRLF
23.     // message-body
25.     // 响应
26.     // HTTP-Version Status-Code Reason-Phrase CRLF
27.     // headers CRLF
28.     // message-body
30.     let contents = fs::read_to_string("hello.html").unwrap();
31.     let response = format!("HTTP/1.1 200 OK\r\n\r\n{}", contents);
33.     stream.write(response.as_bytes()).unwrap();
34.     stream.flush().unwrap();
35. }

复制代码

修改四：

1. use std::fs;
2. use std::io::prelude::*;
3. use std::net::TcpListener;
4. use std::net::TcpStream;
6. fn main() {
7.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
8.     for stream in listener.incoming() {
9.         let stream = stream.unwrap();
11.         handle_connection(stream);
12.     }
13. }
15. fn handle_connection(mut stream: TcpStream) {
16.     let mut buffer = [0; 512];
18.     stream.read(&mut buffer).unwrap();
20.     let get = b"GET / HTTP/1.1\r\n";
22.     if buffer.starts_with(get) {
23.         let contents = fs::read_to_string("hello.html").unwrap();
24.         let response = format!("HTTP/1.1 200 OK\r\n\r\n{}", contents);
26.         stream.write(response.as_bytes()).unwrap();
27.         stream.flush().unwrap();
28.     } else {
29.         let status_line = "HTTP/1.1 404 NOT FOUND\r\n\r\n";
30.         let contents = fs::read_to_string("404.html").unwrap();
32.         let response = format!("{}{}", status_line, contents);
34.         stream.write(response.as_bytes()).unwrap();
35.         stream.flush().unwrap();
36.     }
37. }

复制代码

修改五：

1. use std::fs;
2. use std::io::prelude::*;
3. use std::net::TcpListener;
4. use std::net::TcpStream;
6. fn main() {
7.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
8.     for stream in listener.incoming() {
9.         let stream = stream.unwrap();
11.         handle_connection(stream);
12.     }
13. }
15. fn handle_connection(mut stream: TcpStream) {
16.     let mut buffer = [0; 512];
18.     stream.read(&mut buffer).unwrap();
20.     let get = b"GET / HTTP/1.1\r\n";
22.     let (status_line, filename) = if buffer.starts_with(get) {
23.         ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
24.     } else {
25.         ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
26.     };
28.     let contents = fs::read_to_string(filename).unwrap();
30.     let response = format!("{}{}", status_line, contents);
32.     stream.write(response.as_bytes()).unwrap();
33.     stream.flush().unwrap();
34. }

复制代码

hello.html 文件

1. <!DOCTYPE html>
2. <html lang="en">
3.     <head>
4.         <meta charset="utf-8">
5.         <title>Hello</title>
6.     </head>
8.     <body>
9.         <h1>Hello</h1>
10.         <p>Hi from Rust</p>
11.     </body>
12. </html>

复制代码

404.html 文件

1. <!DOCTYPE html>
2. <html lang="en">
3.     <head>
4.         <meta charset="utf-8">
5.         <title>Hello!</title>
6.     </head>
7.     <body>
8.         <h1>Oops!</h1>
9.         <p>Sorry, I don't know what you're asking for.</p>
10.     </body>
11. </html>

复制代码

单线程Web服务器

1. use std::fs;
2. use std::thread;
3. use std::io::prelude::*;
4. use std::net::TcpListener;
5. use std::net::TcpStream;
6. use std::time::Duration;
8. fn main() {
9.     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
10.     for stream in listener.incoming() {
11.         let stream = stream.unwrap();
13.         handle_connection(stream);
14.     }
15. }
17. fn handle_connection(mut stream: TcpStream) {
18.     let mut buffer = [0; 512];
20.     stream.read(&mut buffer).unwrap();
22.     let get = b"GET / HTTP/1.1\r\n";
23.     let sleep = b"GET /sleep HTTP/1.1\r\n";
25.     let (status_line, filename) = if buffer.starts_with(get) {
26.         ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
27.     } else if buffer.starts_with(sleep) {
28.         thread::sleep(Duration::from_secs(5));
29.         ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
30.     } else {
31.         ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
32.     };
34.     let contents = fs::read_to_string(filename).unwrap();
36.     let response = format!("{}{}", status_line, contents);
38.     stream.write(response.as_bytes()).unwrap();
39.     stream.flush().unwrap();
40. }

复制代码

开启线程

1. use std::fs;
2. use std::thread;
3. use std::io::prelude::*;
4. use std::net::TcpListener;
5. use std::net::TcpStream;
6. use std::time::Duration;
8. fn main() {
9.     let listener = TcpListener::bind("localhost:7878").unwrap();
10.     for stream in listener.incoming() {
11.         let stream = stream.unwrap();
13.         thread::spawn(|| {
14.             handle_connection(stream);
15.         });
16.     }
17. }
19. fn handle_connection(mut stream: TcpStream) {
20.     let mut buffer = [0; 512];
22.     stream.read(&mut buffer).unwrap();
24.     let get = b"GET / HTTP/1.1\r\n";
25.     let sleep = b"GET /sleep HTTP/1.1\r\n";
27.     let (status_line, filename) = if buffer.starts_with(get) {
28.         ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
29.     } else if buffer.starts_with(sleep) {
30.         thread::sleep(Duration::from_secs(5));
31.         ("HTTP/1.1 200 OK\r\n\r\n", "hello.html")
32.     } else {
33.         ("HTTP/1.1 404 NOT FOUND\r\n\r\n", "404.html")
34.     };
36.     let contents = fs::read_to_string(filename).unwrap();
38.     let response = format!("{}{}", status_line, contents);
40.     stream.write(response.as_bytes()).unwrap();
41.     stream.flush().unwrap();
42. }

复制代码

lib.rs 文件

1. use std::thread;
3. pub struct ThreadPool {
4.     threads: Vec<thread::JoinHandle<()>>,
5. }
7. impl ThreadPool {
8.     /// Creates a new ThreadPool.
9.     ///
10.     /// The size is the number of threads in the pool.
11.     ///
12.     /// # Panics
13.     ///
14.     /// The `new` function will panic if the size is zero.
15.     pub fn new(size: usize) -> ThreadPool {
16.         assert!(size > 0);
17.         let mut threads = Vec::with_capacity(size);
19.         for _ in 0..size {
20.             // create some threads and store them in the vector
21.         }
22.         ThreadPool { threads }
23.     }
25.     pub fn execute<F>(&self, f: F)
26.     where
27.         F: FnOnce() + Send + 'static,
28.     {
29.     }
30. }

复制代码

lib.rs 修改一

1. use std::thread;
3. pub struct ThreadPool {
4.     workers: Vec<Worker>,
5. }
7. impl ThreadPool {
8.     /// Creates a new ThreadPool.
9.     ///
10.     /// The size is the number of threads in the pool.
11.     ///
12.     /// # Panics
13.     ///
14.     /// The `new` function will panic if the size is zero.
15.     pub fn new(size: usize) -> ThreadPool {
16.         assert!(size > 0);
17.         let mut workers = Vec::with_capacity(size);
19.         for id in 0..size {
20.             workers.push(Worker::new(id));
21.         }
22.         ThreadPool { workers }
23.     }
25.     pub fn execute<F>(&self, f: F)
26.     where
27.         F: FnOnce() + Send + 'static,
28.     {
29.     }
30. }
32. struct Worker {
33.     id: usize,
34.     thread: thread::JoinHandle<()>,
35. }
37. impl Worker {
38.     fn new(id: usize) -> Worker {
39.         let thread = thread::spawn(|| {});
41.         Worker { id, thread }
42.     }
43. }

复制代码

lib.rs 修改二

1. use std::thread;
2. use std::sync::mpsc;
4. pub struct ThreadPool {
5.     workers: Vec<Worker>,
6.     sender: mpsc::Sender<Job>,
7. }
9. struct Job;
10. impl ThreadPool {
11.     /// Creates a new ThreadPool.
12.     ///
13.     /// The size is the number of threads in the pool.
14.     ///
15.     /// # Panics
16.     ///
17.     /// The `new` function will panic if the size is zero.
18.     pub fn new(size: usize) -> ThreadPool {
19.         assert!(size > 0);
20.         let (sender, receiver) = mpsc::channel();
21.         let mut workers = Vec::with_capacity(size);
23.         for id in 0..size {
24.             workers.push(Worker::new(id, receiver));
25.         }
26.         ThreadPool { workers, sender }
27.     }
29.     pub fn execute<F>(&self, f: F)
30.     where
31.         F: FnOnce() + Send + 'static,
32.     {
33.     }
34. }
36. struct Worker {
37.     id: usize,
38.     thread: thread::JoinHandle<()>,
39. }
41. impl Worker {
42.     fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
43.         let thread = thread::spawn(|| {
44.             receiver;
45.         });
47.         Worker { id, thread }
48.     }
49. }

复制代码

lib.rs 修改三

1. use std::thread;
2. use std::sync::mpsc;
3. use std::sync::Arc;
4. use std::sync::Mutex;
6. pub struct ThreadPool {
7.     workers: Vec<Worker>,
8.     sender: mpsc::Sender<Job>,
9. }
11. struct Job;
12. impl ThreadPool {
13.     /// Creates a new ThreadPool.
14.     ///
15.     /// The size is the number of threads in the pool.
16.     ///
17.     /// # Panics
18.     ///
19.     /// The `new` function will panic if the size is zero.
20.     pub fn new(size: usize) -> ThreadPool {
21.         assert!(size > 0);
22.         let (sender, receiver) = mpsc::channel();
23.         let receiver = Arc::new(Mutex::new(receiver));
24.         let mut workers = Vec::with_capacity(size);
26.         for id in 0..size {
27.             workers.push(Worker::new(id, Arc::clone(&receiver)));
28.         }
29.         ThreadPool { workers, sender }
30.     }
32.     pub fn execute<F>(&self, f: F)
33.     where
34.         F: FnOnce() + Send + 'static,
35.     {
36.     }
37. }
39. struct Worker {
40.     id: usize,
41.     thread: thread::JoinHandle<()>,
42. }
44. impl Worker {
45.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
46.         let thread = thread::spawn(|| {
47.             receiver;
48.         });
50.         Worker { id, thread }
51.     }
52. }

复制代码

lib.rs 修改四

1. use std::thread;
2. use std::sync::mpsc;
3. use std::sync::Arc;
4. use std::sync::Mutex;
6. pub struct ThreadPool {
7.     workers: Vec<Worker>,
8.     sender: mpsc::Sender<Job>,
9. }
11. // struct Job;
12. type Job = Box<FnOnce() + Send + 'static>;
13. impl ThreadPool {
14.     /// Creates a new ThreadPool.
15.     ///
16.     /// The size is the number of threads in the pool.
17.     ///
18.     /// # Panics
19.     ///
20.     /// The `new` function will panic if the size is zero.
21.     pub fn new(size: usize) -> ThreadPool {
22.         assert!(size > 0);
23.         let (sender, receiver) = mpsc::channel();
24.         let receiver = Arc::new(Mutex::new(receiver));
25.         let mut workers = Vec::with_capacity(size);
27.         for id in 0..size {
28.             workers.push(Worker::new(id, Arc::clone(&receiver)));
29.         }
30.         ThreadPool { workers, sender }
31.     }
33.     pub fn execute<F>(&self, f: F)
34.     where
35.         F: FnOnce() + Send + 'static,
36.     {
37.         let job = Box::new(f);
38.         self.sender.send(job).unwrap();
39.     }
40. }
42. struct Worker {
43.     id: usize,
44.     thread: thread::JoinHandle<()>,
45. }
47. impl Worker {
48.     fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
49.         let thread = thread::spawn(move || loop {
50.             let job = receiver.lock().unwrap().recv().unwrap();
52.             println!("Worker {} got a job; executing.", id);
54.             (*job)();
55.         });
57.         Worker { id, thread }
58.     }
59. }

复制代码

lib.rs 修改五
[code]use std::thread;use std::sync::mpsc;use std::sync::Arc;use std::sync::Mutex;pub struct ThreadPool {    workers: Vec,    sender: mpsc::Sender,}// struct Job;// type Job = Box;impl ThreadPool {    /// Creates a new ThreadPool.    ///    /// The size is the number of threads in the pool.    ///    /// # Panics    ///    /// 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));        let mut workers = Vec::with_capacity(size);        for id in 0..size {            workers.push(Worker::new(id, Arc::clone(&receiver)));        }        ThreadPool { workers, sender }    }    pub fn execute(&self, f: F)    where        F: FnOnce() + Send + 'static,    {        let job = Box::new(f);        self.sender.send(job).unwrap();    }}struct Worker {    id: usize,    thread: thread::JoinHandle,}trait FnBox {    fn call_box(self: Box);}impl FnBox for F {    fn call_box(self: Box) {        (*self)()    }}type Job = Box;impl ThreadPool {    /// Creates a new ThreadPool.    ///    /// The size is the number of threads in the pool.    ///    /// # Panics    ///    /// 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));        let mut workers = Vec::with_capacity(size);        for id in 0..size {            workers.push(Worker::new(id, Arc::clone(&receiver)));        }        ThreadPool { workers, sender }    }    pub fn execute(&self, f: F)    where        F: FnOnce() + Send + 'static,    {        let job = Box::new(f);        self.sender.send(job).unwrap();    }}impl Drop for ThreadPool {    fn drop(&mut self) {        for worker in &mut self.workers {            println!("Shutting down worker {}", worker.id);            worker.thread.join().unwrap()        }    }}struct Worker {    id: usize,    thread: thread::JoinHandle,}trait FnBox {    fn call_box(self: Box);}impl FnBox for F {    fn call_box(self: Box) {        (*self)()    }}type Job = Box;impl ThreadPool {    /// Creates a new ThreadPool.    ///    /// The size is the number of threads in the pool.    ///    /// # Panics    ///    /// 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));        let mut workers = Vec::with_capacity(size);        for id in 0..size {            workers.push(Worker::new(id, Arc::clone(&receiver)));        }        ThreadPool { workers, sender }    }    pub fn execute(&self, f: F)    where        F: FnOnce() + Send + 'static,    {        let job = Box::new(f);        self.sender.send(job).unwrap();    }}impl Drop for ThreadPool {    fn drop(&mut self) {        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,    thread: Option,}trait FnBox {    fn call_box(self: Box);}impl FnBox for F {    fn call_box(self: Box) {        (*self)()    }}type Job = Box;impl ThreadPool {    /// Creates a new ThreadPool.    ///    /// The size is the number of threads in the pool.    ///    /// # Panics    ///    /// 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));        let mut workers = Vec::with_capacity(size);        for id in 0..size {            workers.push(Worker::new(id, Arc::clone(&receiver)));        }        ThreadPool { workers, sender }    }    pub fn execute(&self, f: F)    where        F: FnOnce() + Send + 'static,    {        let job = Box::new(f);        self.sender.send(Message::NewJob(job)).unwrap();    }}impl Drop for ThreadPool {    fn drop(&mut self) {        println!("Sending terminate message to all workers.");        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,    thread: Option,}trait FnBox {    fn call_box(self: Box);}impl FnBox for F {    fn call_box(self: Box) {        (*self)()    }}type Job = Box