rust 创建多线程web server

创建一个 http server，处理 http 请求。

创建一个单线程的 web 服务

web server 中主要的两个协议是 http 和 tcp。tcp 是底层协议，http 是构建在 tcp 之上的。

通过std::net库创建一个 tcp 连接的监听对象，监听地址为127.0.0.1:8080.

use std::net::TcpListener;fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();for stream in listener.incoming() {let stream = stream.unwrap();println!("connected!");}
}

运行cargo run，在浏览器中访问http://127.0.0.1:8080,可以看到控制台输出。

浏览器中显示链接被重置，无法被访问，因为没有响应任何数据。通过listener.incoming()方法返回一个迭代器，它是客户端与服务端之间打开的连接。称之为stream流，可以用来处理请求、响应。

首先处理请求，需要读取请求的参数，通过std::io库处理流信息，引入std::io::prelude::*包含一些读写流需要的特定 trait。

use std::io::{prelude::*, BufReader};
use std::net::TcpListener;fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();for stream in listener.incoming() {let mut stream = stream.unwrap();// 处理请求let buf_reader = BufReader::new(&stream);let http_request: Vec<_> = buf_reader.lines().map(|result| result.unwrap()).take_while(|line| !line.is_empty()).collect();println!("requrest:{:#?}", http_request);}
}

BufReader 实现了BufReadtrait,提供了lines方法，通过换行符切割数据流返回一个Result<String,std::io::Error>迭代器。通过map获取到每一个结果值，take_while处理值直到为空结束，然后collect收集结果值。

http_request必须指定类型Vec<_>来收集。在闭包那一节中，迭代器适配器，必须调用消费适配器获取结果。

request-info.png

现在尝试给请求作出一个响应，响应状态码200表示成功响应。一个简单的响应头包括了协议、协议版本、响应状态、状态语句。

let res = "HTTP/1.1 200 OK\r\n\r\n";stream.write_all(res.as_bytes()).unwrap();

重新启动，再次浏览器访问地址，可以看到空白页面，F12查看网络请求，可以看到请求成功

server-200.png

可以增加请求路径http://127.0.0.1:8080/home或增加参数看看请求信息的不同。将请求处理、响应处理放到一个函数中handle_request

接着可以返回一个html文件，这样页面就有了基础的展示效果。新建一个index.html文件

<!DOCTYPE html>
<html lang="en"><head><meta charset="UTF-8" /><meta name="viewport" content="width=<device-width>, initial-scale=1.0" /><title>Document</title></head><body><p>hello world</p></body>
</html>

读取index.html文件，并将文件内容作为响应返回

let res_status = "HTTP/1.1 200 OK\r\n";let contents = fs::read_to_string("index.html").unwrap();let len = contents.len();let res = format!("{res_status}Content-Length:{len}\r\n\r\n{contents}");stream.write_all(res.as_bytes()).unwrap();

再次运行，浏览器访问可以看到页面上已经展示信息。现在只要是所有的请求访问都会返回index.html文件，通常我们会根据访问路径来处理响应，比如http://127.0.0.1:8080/home

限制如果有请求路径或者是参数，则响应一个404.html页面,获取http_request第一个元素匹配GET / HTTP/1.1,响应 200，其他访问都是返回 404.

fn handle_request(mut stream: TcpStream) {// 处理请求let buf_reader = BufReader::new(&stream);let http_request: Vec<_> = buf_reader.lines().map(|result| result.unwrap()).take_while(|line| !line.is_empty()).collect();if http_request[0] == "GET / HTTP/1.1" {let res_status = "HTTP/1.1 200 OK\r\n";let contents = fs::read_to_string("index.html").unwrap();let len = contents.len();let res = format!("{res_status}Content-Length:{len}\r\n\r\n{contents}");stream.write_all(res.as_bytes()).unwrap();} else {let res_status = "HTTP/1.1 404 NOT FOUND\r\n";let contents = fs::read_to_string("404.html").unwrap();let len = contents.len();let res = format!("{res_status}Content-Length:{len}\r\n\r\n{contents}");stream.write_all(res.as_bytes()).unwrap();}
}

优化一下if else里的代码,只有响应状态、响应的文件不一样，其他逻辑都一样。

let (res_status, file_name) = if http_request[0] == "GET / HTTP/1.1" {("HTTP/1.1 200 OK\r\n", "index.html")
} else {("HTTP/1.1 404 NOT FOUND\r\n", "404.html")
};
let contents = fs::read_to_string(file_name).unwrap();let len = contents.len();let res = format!("{res_status}Content-Length:{len}\r\n\r\n{contents}");stream.write_all(res.as_bytes()).unwrap();

main方法中的简化，调用处理请求的函数。

fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();for stream in listener.incoming() {let mut stream = stream.unwrap();// 处理请求handle_request(stream);}
}

现在一个简易的 web 服务就好了，可以处理请求、可以处理响应。在这过程出现的错误我们都用unwrap方法处理，只要遇到错误，直接停止程序，而在真实环境中，需要处理这些错误，避免程序的不可访问。

创建多线程 server 服务

已经构建了单线程的服务，但是它每次只能处理一个请求，只要完成上一个请求之后才能处理下一个连接。如果请求很多，则需要等待，这种表现使得服务性能很差。

首先，来模拟演示一下单线程的堵塞行为，通过线程休眠模拟慢请求

use std::thread::{self};
use std::time::Duration;fn handle_request(mut stream: TcpStream) {// ...// 将if部分改为match匹配，增加/sleep 路径匹配，用以堵塞线程let (res_status, file_name) = match &http_request[0][..] {"GET / HTTP/1.1" => ("HTTP/1.1 200 OK\r\n", "index.html"),"GET /sleep HTTP/1.1" => {// 线程堵塞5sthread::sleep(Duration::from_secs(5));("HTTP/1.1 200 OK", "index.html")}_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),};// ...
}

然后我们打开两个浏览器的 tab 页，访问不同的地址带路径/sleep和不带路径/的，先访问带路径的，可以看到浏览器正在加载，再访问不带路径的也发现浏览器正在加载。等 5 秒过后，全部加载完成，如果直接访问不带路径/则瞬间访问成功。

为了处理这种情况，我们尝试为每一个请求都分配一个线程独立去处理请求任务。

构建一个线程池，当程序每收到新请求时，分配一个线程去处理该请求；其余线程等待处理其他接收到的请求，当线程处理完请求后，返回到线程池等待处理新的请求。这样我们就可以并发处理请求，这样就是服务的吞吐量。

线程池的线程数不易过多，以固有数量的线程等待处理请求。这可以防止拒绝式服务攻击DOS

除了多线程处理服务，还有其他方法改善服务吞吐量，fork/join模型、单线程异步 I/O 模型、多线程异步 I/O 模型。

修改main方法，thread::spawn会创建一个新线程并运行闭包里的代码。

fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();for stream in listener.incoming() {let stream = stream.unwrap();thread::spawn(|| {handle_request(stream);});// handle_request(stream);}
}

现在可以再次尝试请求/sleep和/，可以发现/瞬间就响应了，/sleep还需要等待 5s。如果有上千、上万个请求，我们就要开同等数量的线程，在占用完所有资源后，就会使系统奔溃。

通过线程池，创建有限的线程数量。在处理请求时，内部执行的方法execute会检测空闲的线程并执行之后的请求任务，如果请求超过线程池线程数量，则排队等待。

fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();// 创建线程池let threadPool = ThreadPool::new(4);for stream in listener.incoming() {let stream = stream.unwrap();// thread::spawn(|| {//     handle_request(stream);// });threadPool.execute(|| {handle_request(stream);})// handle_request(stream);}
}

实现ThreadPool线程池类型

ThreadPool 类型并不存在于 rust 库中，需要我们自己实现ThreadPool。

在rust-lib项目中，新建库thread_pool, 在src/lib.rs中,通过new函数实现创建ThreadPool实例，它接受一个参数size为线程的数量；通过定义execute函数接受一个闭包参数，闭包作为参数可以使用三个不同的 traitFn\FnMut\FnOnce,要决定用哪个取决于最终的调用，最终是要调用thread::spawn()的，它是使用了FnOnce的，还需要Send来将闭包从一个线程转移到另一个线程，绑定生命周期'static是因为不知道线程会执行多久。

pub struct ThreadPool;impl ThreadPool {/// 创建线程池////// 线程池中线程的数量pub fn new(size: usize) -> ThreadPool {ThreadPool}pub fn execute<F>(&self, f: F)whereF: FnOnce() + Send + 'static,{}
}

定义完之后，回到项目rust-web项目，引入依赖,在Cargo.toml,

[dependencies]
thread_pool = {path="../rust-lib/thread_pool"}

然后在src/main.rs使用依赖use thread_pool::ThreadPool;, 运行程序cargo run，没有报错正常运行。

在new方法中要保证初始化的线程数是一个有效的值，即size不能为分数或等于 0.这没有意义。然后初始化 vector 实例来存储线程实例,thread::spawn()执行后返回的类型为thread::JoinHandle，它可以管理并等待创建的线程完成任务。

use std::thread;pub struct ThreadPool {threads: Vec<thread::JoinHandle<()>>,
}impl ThreadPool {/// 创建线程池////// 线程池中线程的数量////// # Panics////// `new`函数在size为0 时panicthread_poolpub fn new(size: usize) -> ThreadPool {assert!(size > 0);let mut threads = Vec::with_capacity(size);for _ in 0..size {// 创建对应数量的线程，并把它们存储到vec中}ThreadPool { threads }}// ...
}

之前一直在使用thread::spawn()来创建线程，并执行任务。现在在线程池中，需要提前创建线程，等待任务传入后再执行。标准的 rust 库中没有这样的定义，仍需要自己实现，可以称之为Worker数据结构，这样我们在ThreadPool存储的是Worker实例，在 worker 实例中存储一个单独的JoinHandle<()>实例，并赋予该实例一个唯一的id,方便日志和调用栈区分。

同样的，在ThreadPoolsrc/lib.rs 定义结构体Worker类型,对于外部 worker 类型是私有的，不需要pub定义。

use std::thread;pub struct ThreadPool {// threads: Vec<thread::JoinHandle<()>>,workers: Vec<Worker>,
}
impl ThreadPool {/// 创建线程池////// 线程池中线程的数量pub fn new(size: usize) -> ThreadPool {assert!(size > 0);// let mut threads = Vec::with_capacity(size);let mut workers = Vec::with_capacity(size);for id in 0..size {// 创建对应数量的线程，并把它们存储到vec中workers.push(Worker::new(id))}ThreadPool { workers }}// ...
}struct Worker {id: usize,thread: thread::JoinHandle<()>,
}impl Worker {fn new(id: usize) -> Worker {let thread = thread::spawn(|| {});Worker { id, thread }}
}

运行我们的代码，正常运行。现在需要解决的是向创建的线程传递要处理的请求任务，通过之前文章中学过的channel信道来传递信息.

在ThreadPool中存在一个信道实例充当发送者；并新建一个Job结构体存放用于向信道发送的闭包；execute方法会发送期望执行的任务。

use std::{sync::mpsc, thread};pub struct ThreadPool {// threads: Vec<thread::JoinHandle<()>>,workers: Vec<Worker>,sender: mpsc::Sender<Job>,
}struct Job;impl ThreadPool {pub fn new(size: usize) -> ThreadPool {assert!(size > 0);// let mut threads = Vec::with_capacity(size);let mut workers = Vec::with_capacity(size);// 创建信道实例，提供一个发送者、接收者let (sender, receiver) = mpsc::channel();for id in 0..size {// 创建对应数量的线程，并把它们存储到vec中workers.push(Worker::new(id,receiver))}ThreadPool { workers, sender }}
}

ThreadPool实例存储信道发送者对象sender,需要将接受者实例receiver传递给Worker用于接收传递的信息。

impl Worker {fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {let thread = thread::spawn(|| {receiver;});Worker { id, thread }}
}

这会有一个错误信息，因为在 rust 中信道实现是多生产者、单消费者，不能将receiver接受者传递多个 work 实例。我们希望有一个任务列表，每个任务只允许处理一次。这在之前的文章中

rust 自动化测试、迭代器与闭包、智能指针、无畏并发

已经解决过在线程间共享状态，通过线程安全智能指针Arc<Mutex<T>>，多个线程共享所有权并允许线程修改其值。Arc使得多个 worker 拥有接受端，而Mutex确保一次只有一个 worker 能接收到任务。

use std::{sync::{mpsc, Arc, Mutex},thread,
};impl ThreadPool {pub fn new(size: usize) -> ThreadPool {assert!(size > 0);// let mut threads = Vec::with_capacity(size);let mut workers = Vec::with_capacity(size);let (sender, receiver) = mpsc::channel();// 通过`Arc<T>`创建多所有者,Mutex<T>共享数据let receiver = Arc::new(Mutex::new(receiver));for id in 0..size {// 创建对应数量的线程，并把它们存储到vec中workers.push(Worker::new(id, Arc::clone(&receiver)))}ThreadPool { workers, sender }}
}impl Worker {fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {// ...}
}

最后处理execute方法，它接受的闭包需要分配给空闲的线程并执行，修改Job结构体，它不是一个结构体，是接受execute方法接受的闭包类型的类型别名。

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

在execute方法被调用后，新建Job实例，将任务从信道发送端发出，因为发送可能会失败，所以需要unwrap处理错误的发生。

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

继续优化接受端执行任务的逻辑，在接收到任务后，通过lock获取互斥器来锁定资源，防止其他地方使用资源。通过unwrap处理错误时的情况，在获取了互斥器锁定了资源后，调用recv()方法接受任务Job，这会阻塞当前线程，所有如果当前线程没有任务，则会一直等待直到有用的任务。Mutex<T>可以确保一次只有一个 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!("开始执行任务{id}");job();});Worker { id, thread }}
}

通过loop循环执行闭包，一直向信道的接受端请求任务，并在得到任务时执行它们。

现在执行cargo run,并在浏览器中打开多个 tab 请求地址，可以看到打印输出

multi-web-service.png

不能使用其他循环,比如while let \ if let \ match是因为它们循环时相关的代码块结束都不会丢弃临时值，导致锁守护的资源不能释放，不能被访问。

程序停止与清理

当我们终止程序后，如何去处理未执行完的任务，如何清理资源。

为ThreadPool实现Drop,当线程池被丢弃时，应该join所有线程以确保任务完成。

impl Drop for ThreadPool {fn drop(&mut self) {for worker in &mut self.workers {println!("stop worker {}", worker.id);worker.thread.join().unwrap();}}
}

这里会有一个错误，不能编译，提示没有 worker 所有权，因为我们只得到了一个可变借用，不能调用join来消费线程。通过修改来使得thread实例成为一个Option值，这样就可以通过take方法来获取到其中Some成员值进行处理。清理时可以直接将thread赋值为None

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

通过 rust 代码检测提示信息来修改其他需要调整的地方。Workernew 方法创建实例时，接收thread使用Some(thread)

在停止程序，清理时，通过take()获取到成员值后，再调用join()方法等待线程执行结束。

impl Drop for ThreadPool {fn drop(&mut self) {for worker in &mut self.workers {println!("stop worker {}", worker.id);if let Some(thread) = worker.thread.take() {thread.join().unwrap();}}}
}

正常逻辑来说调用了join()之后会关闭线程，但是由于之前的线程逻辑是循环闭包调用等待接受任务，也就是会导致线程一直不会执行完毕，导致阻塞。一直阻塞在第一个线程结束上。

通过修改ThreadPool的Drop方法来显式丢弃sender。为了转移sender所有权，同样的使用Option类型来传递

pub struct ThreadPool {// threads: Vec<thread::JoinHandle<()>>,workers: Vec<Worker>,sender: Option<mpsc::Sender<Job>>,
}impl ThreadPool {pub fn new(size: usize) -> ThreadPool {// ...// ...ThreadPool {workers,sender: Some(sender),}}pub fn execute<F>(&self, f: F)whereF: FnOnce() + Send + 'static,{let job = Box::new(f);//  self.sender.send(job).unwrap();self.sender.as_ref().unwrap().send(job).unwrap();}
}impl Drop for ThreadPool {fn drop(&mut self) {// 显示的丢弃senderdrop(self.sender.take());for worker in &mut self.workers {println!("stop worker {}", worker.id);if let Some(thread) = worker.thread.take() {thread.join().unwrap();}}}
}

Drop()方法调用显示的丢弃sender后，这会关闭信道，表明了后续不会有消息发送，这时在Worker中无限循环调用接受消息的方法都会返回错误，此时可以修改逻辑在遭遇错误后退出循环。

impl Worker {fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {let thread = thread::spawn(move || loop {let message = receiver.lock().unwrap().recv();match message {Ok(job) => {println!("开始执行任务{id}");job();}Err(_) => {println!("worker {id} disconnected");break;}}});Worker {id,thread: Some(thread),}}
}

现在可以正常清理、停机了，如果希望在服务停止前再处理几个请求，通过take()方法模拟只两个请求进行处理，来验证停机的逻辑。它是Iteratortrait

fn main() {let listener = TcpListener::bind("127.0.0.1:8080").unwrap();// 创建线程池let pool = ThreadPool::new(4);for stream in listener.incoming().take(2) {let stream = stream.unwrap();pool.execute(|| {handle_request(stream);})}
}

现在运行程序cargo run，同时在浏览器请求三次，看看控制台如何打印信息，第三个请求不会被执行。

multi-web-service-stop.png

可以看到只执行完了两次请求，在第一次请求处理完成后，调用了Drop方法显示的丢弃了信道发送者sender,这样整个就导致所有 worker 关闭连接。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/152026.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

温湿度计传感器DHT11控制数码管显示verilog代码及视频

温湿度计传感器DHT11控制数码管显示verilog代码及视频

名称&#xff1a;温湿度计传感器DHT11控制数码管显示 软件&#xff1a;QuartusII 语言&#xff1a;Verilog 代码功能&#xff1a; 使用温湿度传感器DHT11采集环境的温度和湿度&#xff0c;并在数码管显示 本代码已在开发板验证 开发板资料&#xff1a; 大西瓜第一代FPGA升级…
阅读更多...
Vue--》简易资金管理系统后台项目实战（前端）

Vue--》简易资金管理系统后台项目实战（前端）

今天开始使用 vue3 + ts + node 搭建一个简易资金管理系统的前后端分离项目，因为前后端分离所以会分两个专栏分别讲解前端与后端的实现，后端项目文章讲解可参考：后端链接，我会在前后端的两类专栏的最后一篇文章中会将项目代码开源到我的github上，大家可以自行去进行下载运…
阅读更多...
Notepad++安装插件和配置快捷键

Notepad++安装插件和配置快捷键

Notepad是一款轻量级、开源的文件编辑工具&#xff0c;可以编辑、浏览文本文件、二进制文件、.cpp、.java、*.cs等文件。Notepad每隔1个月&#xff0c;就有一个新版本&#xff0c;其官网是&#xff1a; https://github.com/notepad-plus-plus/notepad-plus-plus。这里介绍其插件…
阅读更多...
系列十四、Spring如何处理线程安全问题

系列十四、Spring如何处理线程安全问题

一、线程安全问题出现的原因 Spring中出现线程安全的原因是&#xff0c;单实例bean中存在成员变量&#xff0c;并且有对这个bean进行读写的操作&#xff0c;因此出现了线程安全的问题。 二、案例代码 2.1、MySpringConfig /*** Author : 一叶浮萍归大海* Date: 2023/10/24 1…
阅读更多...
【C++】详解map和set基本接口及使用

【C++】详解map和set基本接口及使用

文章目录 一、关联式容器与键值对1.1关联式容器&#xff08;之前学的都是序列容器&#xff09;1.2键值对pairmake_pair函数&#xff08;map在插入的时候会很方便&#xff09; 1.3树形结构的关联式容器 二、set2.1set的基本介绍2.1默认构造、迭代器区间构造、拷贝构造&#xff0…
阅读更多...
Mysql数据库 4.SQL语言 DQL数据查询语言 查询

Mysql数据库 4.SQL语言 DQL数据查询语言 查询

DQL数据查询语言 从数据表中提取满足特定条件的记录 1.单表查询 2.多表查询 查询基础语法 select 关键字后指定要查询到的记录的哪些列 语法&#xff1a;select 列名&#xff08;字段名&#xff09;/某几列/全部列 from 表名 [具体条件]&#xff1b; select colnumName…
阅读更多...
Python环境下LaTeX数学公式转图像方案调研与探讨

Python环境下LaTeX数学公式转图像方案调研与探讨

目录 引言方案一&#xff1a;基于LaTeX环境方案二&#xff1a;基于KaTeX(推荐) 方案三&#xff1a;基于Matplotlib写在最后 引言 近来&#xff0c;涉及到一些公式识别的项目&#xff0c;输入是公式的图像&#xff0c;输出是LaTeX格式的数学公式字符串。 这类项目一般都采用深…
阅读更多...
如何隐藏woocommerce 后台header，woocommerce-layout__header

如何隐藏woocommerce 后台header，woocommerce-layout__header

如何隐藏woocommerce 后台header&#xff0c;woocommerce-layout__header WooCommerce |Products Store Activity| Inbox| Orders| Stock| Reviews| Notices| breadcrumbs 在 functions.php 里添加如下代码即可&#xff1a; // Disable WooCommerce Header in WordPress Admi…
阅读更多...
C++数据结构X篇_21_插入排序（稳定的排序）

C++数据结构X篇_21_插入排序（稳定的排序）

文章目录 1. 插入排序原理2. 算法图解3. 核心代码&#xff1a;4. 插入排序整体代码实现 1. 插入排序原理 插入排序是一种最简单直观的排序算法&#xff0c;它的工作原理是通过构建有序序列&#xff0c;对于未排序数据&#xff0c;在已排序序列中从后向前扫描&#xff0c;找到相…
阅读更多...
matlab创建矩阵、理解三维矩阵

matlab创建矩阵、理解三维矩阵

1.创建矩阵 全0矩阵&#xff1a;a zeros(2,3,4) 全1矩阵&#xff1a;a ones(2,3,4) &#xff01;和python不一样的地方&#xff01;此处相当于创建了4页2行3列的矩阵&#xff0c;而在python里是2页3行4列。 对第1页的第2行第3列元素进行修改&#xff1a;
阅读更多...
【jenkins】centos7在线安装jenkins

【jenkins】centos7在线安装jenkins

一、系统要求 最低推荐配置 256MB可用内存 1GB可用磁盘空间(作为一个Docker容器运行jenkins的话推荐10GB) 软件配置 Java 8—​无论是Java运行时环境&#xff08;JRE&#xff09;还是Java开发工具包&#xff08;JDK&#xff09;都可以 二、安装jenkins 准备一台安装有ce…
阅读更多...
Maven入门与开箱即用

Maven入门与开箱即用

一、初识 Maven&#xff08;了解&#xff09; 1、项目遇到的问题 构建&#xff1a;编译代码&#xff0c;运行测试&#xff0c;打包&#xff0c;部署应用&#xff0c;运行服务器等&#xff1b;依赖&#xff1a;项目依赖大量的第三方包&#xff0c;第三方包又依赖另外的包&…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023