当前位置：首页 > news >正文

rust学习-构建服务器

news 2025/7/7 13:16:40

单线程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中的值，要么Paniclet stream = stream.unwrap();// stream 此时类型：std::net::tcp::TcpStreamhandle_connection(stream);}
}fn handle_connection(mut stream: TcpStream) {let mut buffer = [0; 1024]; // [dataType;size]，为每一个元素初始化为0stream.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.9println!("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";// 给前端传输htmllet 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 中的字节开头，返回404let 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>whereF: 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)whereF: 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 会遍历通道的接收端并执行任何接收到的任务。

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)whereF: 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 所提供的通道实现是多生产者，单消费者// 将通道的接收端传递给 workerfn 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)whereF: 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();
}