Rust小项目

基础技术 / rust
Word count: 1.7k   |   Reading time≈ 9 min

结合前面掌握的基础技术，来完成一个小项目

1. 单线程webserver实现

简单的request请求以及response响应

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};
use std::fs;

fn handle_client(mut stream: TcpStream) {
    // 第一种处理请求方式：浏览器请求http://localhost:9999 后台则会打印下面这句话
    // println!("收到了一些内容")

    // 第二种处理请求方式：读取并打印stream接收到的请求
    /*let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();
    println!("Request:{}",String::from_utf8_lossy(&buffer[..]))*/

    // 第三种处理请求方式：响应结果，这里返回http正常的提示
    /*let reponse = "HTTP/1.1 200 OK\r\n\r\n";
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();*/

    // 第四种处理请求方式：响应返回一个网页；浏览器兼容问题，可以使用 curl http://127.0.0.1:9999
    /*let content = fs::read_to_string("main.html").unwrap();
    let reponse = format!("HTTP/1.1 200 OK\r\n\r\n{}", content);
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();*/

    // 第五种处理请求方式：有条件的返回网页；请求http://localhost:9999返回main.html，请求http://localhost:9999/xxx返回404.html
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();
    let get = b"Get / HTTP/1.1/\r\n";  //b表示转化为字节
    //笨重实现
    /*if buffer.starts_with(get) {
        println!("收到了一些内容");
        //返回main.html
        let content = fs::read_to_string("main.html").unwrap();
        let reponse = format!("HTTP/1.1 200 OK\r\n\r\n{}", content);
        stream.write(reponse.as_bytes()).unwrap();
        stream.flush().unwrap();
    } else {
        //返回404
        let content = fs::read_to_string("404.html").unwrap();
        let reponse = format!("HTTP/1.1 404 NOT FOUND\r\n\r\n{}", content);
        stream.write(reponse.as_bytes()).unwrap();
        stream.flush().unwrap();
    }*/
    //代码优化实现，代码更清爽
    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK\r\n\r\n{}", "main.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n{}", "404.html")
    };
    let contents = fs::read_to_string(filename).unwrap();
    let reponse = format!("{}{}",status_line,contents);
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();
}

fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9999")?;
    //?表示Result，等价如下：
    // let listener = TcpListener::bind("127.0.0.1:80").unwrap();

    // accept connections and process them serially
    for stream in listener.incoming() {
        handle_client(stream?);
    }
    Ok(())
}

2. 多线程webserver实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};
use std::fs;
use std::{thread, time};

fn handle_client(mut stream: TcpStream) {
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();
    let get = b"Get / HTTP/1.1/\r\n";  //b表示转化为字节
    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK\r\n\r\n{}", "main.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n{}", "404.html")
    };
    let contents = fs::read_to_string(filename).unwrap();
    let reponse = format!("{}{}", status_line, contents);
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();

    let te = time::Duration::from_millis(10000);
    thread::sleep(te);
}

fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9999")?;
    let mut thread_vec: Vec<thread::JoinHandle<()>> = Vec::new();
    for stream in listener.incoming() {
        let stream = stream.unwrap();
        let handle = thread::spawn(move || {
            handle_client(stream);
        });
        thread_vec.push(handle)
    }
    for handle in thread_vec{
        handle.join().unwrap();
    }
    Ok(())
}

3. 线程池webserver实现-初版

lib中：
cargo new mylib --lib

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
use std::{thread};
use std::sync::mpsc;
use std::sync::Arc;
use std::sync::Mutex;

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{
                let job = receiver.lock().unwrap().recv().unwrap();
                println!("Worker{} got a job!",id);
                job();
            }
            receiver;
        });
        Worker {
            id,
            thread,
        }
    }
}

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

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

impl ThreadPool {
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);
        let mut workers = Vec::with_capacity(size);
        let (sender, receiver) = mpsc::channel();
        let receiver = Arc::new(Mutex::new(receiver)); //线程安全
        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

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

main.rs中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};
use std::fs;
use std::{thread, time};
use mylib::ThreadPool;

fn handle_client(mut stream: TcpStream) {
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();
    let get = b"Get / HTTP/1.1/\r\n";  //b表示转化为字节
    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK\r\n\r\n{}", "main.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n{}", "404.html")
    };
    let contents = fs::read_to_string(filename).unwrap();
    let reponse = format!("{}{}", status_line, contents);
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();

    let te = time::Duration::from_millis(10000);
    thread::sleep(te);
}

fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9999")?;
    let pool = ThreadPool::new(4);
    for stream in listener.incoming() {
        let stream = stream.unwrap();
        pool.execute(||{
            handle_client(stream)
        })
    }

    Ok(())
}

4. 线程池webserver实现-完善

支持线程池结束

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
use std::{thread};
use std::sync::mpsc;
use std::sync::Arc;
use std::sync::Mutex;

struct Worker {
    id: usize,
    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 {} receive a job", id);
                        job();
                    }
                    Message::Terminate => {
                        println!("worker {} receive terminate", id);
                        break;
                    }
                }
            }
        });
        Worker {
            id,
            thread: Some(thread),
        }
    }
}

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

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

enum Message {
    NewJob(Job),
    Terminate,
}

impl ThreadPool {
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);
        let mut workers = Vec::with_capacity(size);
        let (sender, receiver) = mpsc::channel();
        let receiver = Arc::new(Mutex::new(receiver)); //线程安全
        for id in 0..size {
            workers.push(Worker::new(id, Arc::clone(&receiver)));
        }

        ThreadPool {
            workers,
            sender,
        }
    }
    pub fn execute<F>(&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) {
        //发送中断消息
        for _ in &mut self.workers {
            self.sender.send(Message::Terminate).unwrap();
        }
        for worker in &mut self.workers {
            //等待worker结束
            if let Some(thread) = worker.thread.take() {
                thread.join().unwrap();
            }
        }
    }
}

main.rs中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
use std::net::{TcpListener, TcpStream};
use std::io::{Read, Write};
use std::fs;
use std::{thread, time};
use mylib::ThreadPool;

fn handle_client(mut stream: TcpStream) {
    let mut buffer = [0; 512];
    stream.read(&mut buffer).unwrap();
    let get = b"Get / HTTP/1.1/\r\n";  //b表示转化为字节
    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK\r\n\r\n{}", "main.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND\r\n\r\n{}", "404.html")
    };
    let contents = fs::read_to_string(filename).unwrap();
    let reponse = format!("{}{}", status_line, contents);
    stream.write(reponse.as_bytes()).unwrap();
    stream.flush().unwrap();

    let te = time::Duration::from_millis(10000);
    thread::sleep(te);
}

fn main() -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:9999")?;
    let pool = ThreadPool::new(4);
    for stream in listener.incoming().take(4) {
        let stream = stream.unwrap();
        pool.execute(||{
            handle_client(stream)
        })
    }

    Ok(())
}

总结

编辑完毕

  • Copyrights © 2017-2023 Jason
  • Visitors: | Views:

谢谢打赏～

微信