rust学习-构建服务器-Toy模板网

这篇具有很好参考价值的文章主要介绍了rust学习-构建服务器。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

单线程server

服务器会依次处理每一个请求，在完成第一个连接的处理之前不会处理第二个连接

// cat main.rs
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        // stream 此时类型：core::result::Result<std::net::tcp::TcpStream, std::io::error::Error>
        // unwrap() 要么取出Result的Some中的值，要么Panic
        let stream = stream.unwrap();
		// stream 此时类型：std::net::tcp::TcpStream
        handle_connection(stream);
    }
}

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 1024]; // [dataType;size]，为每一个元素初始化为0

    stream.read(&mut buffer).unwrap();
    // 浏览器输入得到内容如下
    // Request: GET /favicon.ico HTTP/1.1
	// Host: 127.0.0.1:7878
	// Connection: keep-alive
	// sec-ch-ua: "Not.A/Brand";v="8", "Chromium";v="114", "Google Chrome";v="114"
	// sec-ch-ua-mobile: ?0
	// User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36
	// sec-ch-ua-platform: "macOS"
	// Accept: image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8
	// Sec-Fetch-Site: same-origin
	// Sec-Fetch-Mode: no-cors
	// Sec-Fetch-Dest: image
	// Referer: http://127.0.0.1:7878/
	// Accept-Encoding: gzip, deflate, br
	// Accept-Language: zh-CN,zh;q=0.9
    println!("Request: {}", String::from_utf8_lossy(&buffer[..]));

	// 一个微型的成功 HTTP 响应，格式如下
	// HTTP-Version Status-Code Reason-Phrase CRLF
	// headers CRLF
	// message-body
    // let response = "HTTP/1.1 200 OK\r\n\r\n";

	// 给前端传输html
	let contents = fs::read_to_string("hello.html").unwrap();
    let response = format!(
        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
        contents.len(),
        contents
    );

	// flush 会等待并阻塞程序执行直到所有字节都被写入连接中
	// TcpStream 包含一个内部缓冲区来最小化对底层操作系统的调用
    stream.write(response.as_bytes()).unwrap();
    stream.flush().unwrap();
}

// 客户端的请求行
// Method Request-URI HTTP-Version CRLF
// headers CRLF
// message-body

// hello.html
<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Hello!</title>
  </head>
  <body>
    <h1>Hello!</h1>
    <p>Hi from Rust</p>
  </body>
</html>

区分响应

// cat main.rs
use std::fs;
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        handle_connection(stream);
    }
}

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 1024];

    stream.read(&mut buffer).unwrap();
    println!("Request: {}", String::from_utf8_lossy(&buffer[..]));

	// 将与 / 请求相关的数据硬编码进变量 get
	// 数据开头增加 b"" 字节字符串语法将其转换为字节字符串
    let get = b"GET / HTTP/1.1\r\n";

	// 检查 buffer 是否以 get 中的字节开头。如果是，这就是一个格式良好的 / 请求
    if buffer.starts_with(get) {
        let contents = fs::read_to_string("hello.html").unwrap();

        let response = format!(
            "HTTP/1.1 200 OK\r\nContent-Length: {}\r\n\r\n{}",
            contents.len(),
            contents
        );

        stream.write(response.as_bytes()).unwrap();
        stream.flush().unwrap();
    } else {
    	// 如果 buffer 不 以 get 中的字节开头，返回404
        let status_line = "HTTP/1.1 404 NOT FOUND";
        let contents = fs::read_to_string("404.html").unwrap();

        let response = format!(
            "{}\r\nContent-Length: {}\r\n\r\n{}",
            status_line,
            contents.len(),
            contents
        );

        stream.write(response.as_bytes()).unwrap();
        stream.flush().unwrap();
    }
}

404.html

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="utf-8">
    <title>Hello!</title>
  </head>
  <body>
    <h1>Oops!</h1>
    <p>Sorry, I don't know what you're asking for.</p>
  </body>

重构 handle_connection

fn handle_connection(mut stream: TcpStream) {
    // --snip--

    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND", "404.html")
    };

    let contents = fs::read_to_string(filename).unwrap();

    let response = format!(
        "{}\r\nContent-Length: {}\r\n\r\n{}",
        status_line,
        contents.len(),
        contents
    );

    stream.write(response.as_bytes()).unwrap();
    stream.flush().unwrap();
}

单线程

use std::thread;
use std::time::Duration;
use std::fs;
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        handle_connection(stream);
    }
}

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 1024];

    stream.read(&mut buffer).unwrap();
    println!("Request: {}", String::from_utf8_lossy(&buffer[..]));

    let get = b"GET / HTTP/1.1\r\n";
    let sleep = b"GET /sleep HTTP/1.1\r\n";

    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK", "hello.html")
    } else if buffer.starts_with(sleep) {
        thread::sleep(Duration::from_secs(5));
        ("HTTP/1.1 200 OK", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND", "404.html")
    };

    let contents = fs::read_to_string(filename).unwrap();
    let response = format!(
        "{}\r\nContent-Length: {}\r\n\r\n{}",
        status_line,
        contents.len(),
        contents
    );

    stream.write(response.as_bytes()).unwrap();
    stream.flush().unwrap();
}

多线程

奔溃版本1

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        thread::spawn(|| {
            handle_connection(stream);
        });
    }
}

//  thread::spawn的类型
// F是参数类型，T是返回类型
// F 的 trait 有 （FnOnce() -> T）、Send，生命周期有  'static
// 处理请求的线程只会执行闭包一次，所以用FnOnce
// 需要 Send 来将闭包从一个线程转移到另一个线程
// 'static 是因为并不知道线程会执行多久
pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where
        F: FnOnce() -> T + Send + 'static,
        T: Send + 'static

根据奔溃版本和thread::spawn构造假想线程池

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
    let pool = ThreadPool::new(4);

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        pool.execute(|| {
            handle_connection(stream);
        });
    }
}


impl ThreadPool {
    // --snip--

    pub fn execute<F>(&self, f: F)
        where
            F: FnOnce() + Send + 'static
    {

    }
}

构造线程池-糟糕版本

thread::spawn，期望获取一些一旦创建线程就应该执行的代码
但是对于线程池不适用，线程池是当在需要时才执行代码执行

（1）定义 Worker 结构体存放 id 和 JoinHandle<()>
（2）修改 ThreadPool 存放一个 Worker 实例的 vector
（3）定义 Worker::new 函数，它获取一个 id 数字并返回一个带有 id 和用空闭包分配的线程的 Worker 实例
（4）在 ThreadPool::new 中，使用 for 循环计数生成 id，使用这个 id 新建 Worker，并储存进 vector 中

execute 将通过 ThreadPool ，向其中空闲的线程 Worker 实例发送任务。
（1）ThreadPool 会创建一个通道并充当发送端。
（2）每个 Worker 将会充当通道的接收端。
（3）新建一个 Job 结构体来存放用于向通道中发送的闭包。
（4）execute 方法会在通道发送端发出期望执行的任务。
（5）在线程中，Worker 会遍历通道的接收端并执行任何接收到的任务。文章来源地址https://www.toymoban.com/news/detail-618927.html

use std::thread;
use std::sync::mpsc;

pub struct ThreadPool {
	workers: Vec<Worker>,
    // threads: Vec<thread::JoinHandle<()>>,

	sender: mpsc::Sender<Job>,
}

struct Job;

impl ThreadPool {
    /// 创建线程池。
    ///
    /// 线程池中线程的数量。
    ///
    /// # Panics
    ///
    /// `new` 函数在 size 为 0 时会 panic。
    pub fn new(size: usize) -> ThreadPool {
       // 创建一个没有任何线程的线程池应该是不可恢复的错误
		assert!(size > 0);

		/*
		// with_capacity, 与 Vec::new 做了同样的工作
		// 它为 vector 预先分配空间
		// 预分配比 Vec::new 要稍微有效率
        let mut threads = Vec::with_capacity(size);
        for _ in 0..size {
            // create some threads and store them in the vector
        }

        ThreadPool {
            threads
        }
		*/

		/*
		let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            workers.push(Worker::new(id));
        }

        ThreadPool {
            workers
        }
		*/

		// 创建通道并让 ThreadPool 实例充当发送端
		// Job是通道的待执行任务
		let (sender, receiver) = mpsc::channel();
        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            workers.push(Worker::new(id, receiver));
        }

        ThreadPool {
            workers,
            sender,
        }
	}

	// 在 execute 方法中获得期望执行的闭包
    pub fn execute<F>(&self, f: F)
        where
            F: FnOnce() + Send + 'static
    {

    }
}

struct Worker {
    id: usize,
    thread: thread::JoinHandle<()>,
}

/*
impl Worker {
    fn new(id: usize) -> Worker {
        // 一个空闭包
		let thread = thread::spawn(|| {});

        Worker {
            id,
            thread,
        }
    }
}
*/

impl Worker {
	// Rust 所提供的通道实现是多生产者，单消费者
	// 将通道的接收端传递给 worker
    fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
        let thread = thread::spawn(|| {
            receiver;
        });

        Worker {
            id,
            thread,
        }
    }
}

终极版本

// cat lib.rs
use std::thread;
use std::sync::mpsc;
use std::sync::Arc;
use std::sync::Mutex;

pub struct ThreadPool {
	workers: Vec<Worker>,
    // threads: Vec<thread::JoinHandle<()>>,

	sender: mpsc::Sender<Job>,
}

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

impl ThreadPool {
    /// 创建线程池。
    ///
    /// 线程池中线程的数量。
    ///
    /// # Panics
    ///
    /// `new` 函数在 size 为 0 时会 panic。
    pub fn new(size: usize) -> ThreadPool {
       // 创建一个没有任何线程的线程池应该是不可恢复的错误
		assert!(size > 0);

		/*
		// with_capacity, 与 Vec::new 做了同样的工作
		// 它为 vector 预先分配空间
		// 预分配比 Vec::new 要稍微有效率
        let mut threads = Vec::with_capacity(size);
        for _ in 0..size {
            // create some threads and store them in the vector
        }

        ThreadPool {
            threads
        }
		*/

		/*
		let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            workers.push(Worker::new(id));
        }

        ThreadPool {
            workers
        }
		*/

		// 创建通道并让 ThreadPool 实例充当发送端
		// Job是通道的待执行任务
		/*
		let (sender, receiver) = mpsc::channel();
        let mut workers = Vec::with_capacity(size);
        for id in 0..size {
            workers.push(Worker::new(id));
        }

        ThreadPool {
            workers,
            sender,
        }
		*/

		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,
        }
	}

	// 在 execute 方法中获得期望执行的闭包
    pub fn execute<F>(&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<()>,
}

/*
impl Worker {
    fn new(id: usize) -> Worker {
        // 一个空闭包
		let thread = thread::spawn(|| {});

        Worker {
            id,
            thread,
        }
    }
}
*/

// Rust 所提供的通道实现是多生产者，单消费者
// 将通道的接收端传递给 worker
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,
        }
    }
}

// cat main.rs
use std::thread;
use std::time::Duration;
use std::fs;
use std::io::prelude::*;
use std::net::TcpListener;
use std::net::TcpStream;
use rust_demo::ThreadPool;

fn main() {
    let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
    let pool = ThreadPool::new(4);

    for stream in listener.incoming() {
        let stream = stream.unwrap();

        pool.execute(|| {
            handle_connection(stream);
        });
    }
}

fn handle_connection(mut stream: TcpStream) {
    let mut buffer = [0; 1024];

    stream.read(&mut buffer).unwrap();
    println!("Request: {}", String::from_utf8_lossy(&buffer[..]));

    let get = b"GET / HTTP/1.1\r\n";
    let sleep = b"GET /sleep HTTP/1.1\r\n";

    let (status_line, filename) = if buffer.starts_with(get) {
        ("HTTP/1.1 200 OK", "hello.html")
    } else if buffer.starts_with(sleep) {
        thread::sleep(Duration::from_secs(5));
        ("HTTP/1.1 200 OK", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND", "404.html")
    };

    let contents = fs::read_to_string(filename).unwrap();
    let response = format!(
        "{}\r\nContent-Length: {}\r\n\r\n{}",
        status_line,
        contents.len(),
        contents
    );

    stream.write(response.as_bytes()).unwrap();
    stream.flush().unwrap();
}

到了这里，关于rust学习-构建服务器的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！