前言#
昨天我们了解了下什么是rust的异步生态，也知道了一些比较热门的异步生态
今天也差不多学完了（除了还没有完善TODO状态的），那么这一节就是来做一个项目。
另外如果不想手动的话直接copy这个example里的源码即可：async-book/main.rs at master · rust-lang/async-book (github.com)
基于Async Rust实现一个并发web项目#
之前我们在学入门文档的时候最终也是写了一个基于thread OS的并发web server，那么我们其实可以基于那个项目改成async的。
Final Project: Building a Multithreaded Web Server - The Rust Programming Language (rust-lang.org)
那么我们之前放到github上的demo就派上用场了。
git clone https://github.com/1714080902120/rust_web_demo.git 选择async runtime#
一般一个HTTP的服务器可以并发地对接很多个客户端，也就是说前面的请求不能堵塞后面的请求。
之前我们是用线程池的方式来缓解这个问题的，那么今天我们就来把线程池改成异步的方式。
不过在这之前，我们需要选择使用哪个async runtime来帮助我们写异步代码。
这里咱就直接使用async-std[3]这个crate了。
既然选定了，那么我们就来加载下吧
cargo add async-std[dependencies.async-std]
version = "1.6"
features = ["attributes"]接着我们来处理下入口
我们得先把之前多线程的东西去掉
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
handle_connection(stream);
}
}去掉线程池之后现在我们的代码变回了单线程代码了，运行之后表现正常。
然后我们来把handle_connection方法变成异步的
async fn handle_connection(mut stream: TcpStream) {
let buf_reader = BufReader::new(&mut stream);
let http_request: Vec<_> = buf_reader
.lines()
.map(|result| result.unwrap())
.take_while(|line| !line.is_empty())
.collect();
let request_line: Vec<_> = http_request
.get(0)
.unwrap()
.split_ascii_whitespace()
.collect();
let path = request_line.get(1).unwrap().to_string();
let (status_line, file_path) = match path.as_str() {
"/" => ("HTTP/1.1 200 OK", "index.html"),
"/sleep" => {
thread::sleep(Duration::from_secs(5));
("HTTP/1.1 200 OK", "index.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let res_body = fs::read_to_string(file_path).unwrap();
let res_headers = format!("Content-Type: html;\r\nContent-Length:{}", res_body.len());
let res = format!("{status_line}\r\n{res_headers}\r\n\r\n{res_body}");
stream.write_all(&res.as_bytes()).unwrap();
} 现在变成异步的了，那么我们要怎么在main函数中await这个异步函数呢？main函数作为入口默认是无法变成异步的。
我们之前的方法是使用block_on的方法来执行异步，我们来改下
use async_std::task::block_on;
fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
block_on(handle_connection(stream));
}
} 这样确实可以，但是不够优雅。我们还是希望可以使用.await的语法。
这个时候async-std就提供了可以允许我们给main写async的方式。
#[async_std::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
for stream in listener.incoming() {
let stream = stream.unwrap();
handle_connection(stream).await;
}
}我们来稍微看下async_std::main这个属性宏的核心源码
可以看到就是改成我们最初的写法而已，我们的async main被一个同步的main套住，而这个同步的main实际上就变成了真正的入口函数，而它里面async main则是我们的代码。
那么初步的异步就完成了，现在是一个单线程异步，由于我们的listener.incoming返回的是同步的iterator，所以这个时候我们await了和没await实际上没差，同样都会堵塞。
我们之前提供了一个/sleep的URI，进入会延迟5秒后再响应，如果请求它的同时去请求其它页面，那么一样会被堵塞，至少5秒后才会响应。
那么我们现在就动手来实现异步并发吧。
实现异步并发#
还记得我们之前学到过的可以支持同时执行多个异步任务的trait吗？
对的，我们可以用Stream这个trait来实现并发。
之前我们了解到Stream和Iterator实际上差不多，区别在于同步和异步。
而async_std这个crate中也提供了用Stream封装的TcpListener：async_std::net::TcpListener。
另外stream都异步了，那么TcpStream自然也得是异步的，这个包中同样提供了这个封装：async_std::net::TcpStream。
我们来改下代码(切忌不要跟着官方的文档给的代码来写，是无法运行的)
use std::{fs, str::from_utf8, time::Duration};
use async_std::{
self,
net::{TcpListener, TcpStream},
prelude::*,
task::{self, block_on},
io::{ BufReader }
};
async fn handle_connection(mut stream: TcpStream) {
let mut buffer = [0; 1024];
stream.read(&mut buffer).await.unwrap();
let req: Vec<&str> = from_utf8(&buffer).unwrap().split("\r\n").filter(|item| { !item.is_empty() }).collect();
let str_list: Vec<&str> = req.get(0).unwrap().split_ascii_whitespace().collect();
let path = str_list.get(1).unwrap();
let (status_line, file_path) = match *path {
"/" => ("HTTP/1.1 200 OK", "index.html"),
"/sleep" => {
task::sleep(Duration::from_secs(5)).await;
("HTTP/1.1 200 OK", "index.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let res_body = fs::read_to_string(file_path).unwrap();
let res_headers = format!("Content-Type: html;\r\nContent-Length:{}", res_body.len());
let res = format!("{status_line}\r\n{res_headers}\r\n\r\n{res_body}");
stream.write(&res.as_bytes()).await.unwrap();
stream.flush().await.unwrap();
}我们使用stream.read方法替换原来的BufRead::new的方法，不过这里其实也可以不改，因为async-std这个crate中也有提供这个BufReader的async trait。
let buf = BufReader::new(&mut stream);
let mut buf = buf.buffer();
let req: Vec<&str> = from_utf8(&buf).unwrap().split("\r\n").filter(|item| { !item.is_empty() }).collect();然后我们使用std::str的from_utf8的方法把buffer转换成&str。
然后再根据\r\n和空格切割拿到URI。
然后在write之后我们还调用了flush这个方法，来看下它的作用。
可以通过注释看出这是用来确保所有的数据都到达它们的目的地。
还有一点，那就是之前的thread::sleep替换成了async_std::task::sleep，这样也就不会堵塞了。
那么这个handle_connection函数我们就处理完了，但是我们的代码还是没法跑的，因为incoming的stream不对劲，不是async的TcpListener。
我们来改下main函数中的代码。
不过在这之前，我们需要安装下futures，因为会使用到这个crate的东西。
[dependencies]
futures = "0.3"或者cargo add futures都行。
use async_std::{
self,
net::{TcpListener, TcpStream},
prelude::*,
task::{self},
};
use futures::stream::StreamExt;
use async_std::task::spawn;
#[async_std::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:7878").await.unwrap();
listener
.incoming()
.for_each_concurrent(/* limit */ None, |stream| async move {
let stream = stream.unwrap();
spawn(handle_connection(stream));
})
.await;
} 这里的两个重点是for_each_concurrent和spawn两个方法
这个for_each_concurrent看名字就知道是用来并发的，它来自于futures::stream::StreamExt，所以
自然就得安装futures这个crate。原理应该类似于join!，都是将它们合并，手动去poll它们让它们进入到队列中，这样就不需要等待前面的ready了。
然后是spawn这个方法，来自于async_std::task
在使用这个spawn方法之前，我们的代码已经可以跑了，并且是并发的。
在请求/sleep的同时请求/会发现/的不需要等待/sleep就加载了
不过现在我们的项目还是单线程的，也就是说同步异步都是跑在同一个线程的，其实我们可以搭配线程来实现多线程异步。
这个async_std::task::spawn的方法是std::thread::spawn方法的异步method，使用它可以把任务派发到独立的线程里执行。
OK，我们的demo现在已经是多线程异步的了。
接下来我们来写测试代码。
测试TCP服务器#
之前使用thread实现的并发服务器并没有编写测试代码，这次算是补上了。
我们准备来写个测试单元来调用handle_connection这个方法模拟请求输入和输出。
不过这就意味着我们需要编写测试数据，在js中相信大家想都不用想，直接就知道怎么写mock数据了，但是在rust中不行，受限于type bound等约束，要去模拟这个数据就意味着数据类型也得保持一致，这就非常繁琐复杂。
在这里，我们需要mock的数据是TcpStream类型的。
而这个类型的bound我们很难去分析全，但是有几个我们是可以确定的：
async_std::io::{Read, Write}Unpin
这几个实际上是我们要用到的，所以我们其实可以把TcpStream简化成impl Read + Write + Unpin
先来改下我们的handle_connection方法
async fn handle_connection(mut stream: impl Read + Write + Unpin) {
let mut buffer = [0; 1024];
stream.read(&mut buffer).await.unwrap();
let req: Vec<&str> = from_utf8(&buffer).unwrap().split("\r\n").filter(|item| { !item.is_empty() }).collect();
let str_list: Vec<&str> = req.get(0).unwrap().split_ascii_whitespace().collect();
let path = str_list.get(1).unwrap();
let (status_line, file_path) = match *path {
"/" => ("HTTP/1.1 200 OK", "index.html"),
"/sleep" => {
task::sleep(Duration::from_secs(5)).await;
("HTTP/1.1 200 OK", "index.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let res_body = fs::read_to_string(file_path).unwrap();
let res_headers = format!("Content-Type: html;\r\nContent-Length:{}", res_body.len());
let res = format!("{status_line}\r\n{res_headers}\r\n\r\n{res_body}");
stream.write(&res.as_bytes()).await.unwrap();
stream.flush().await.unwrap();
}
然后我们就可以来写mock数据了。
不过在这之前，我们需要把handle_connection方法迁移到lib.rs中，这样才好测试。
use std::{
sync::{
mpsc::{self, Receiver},
Arc, Mutex,
},
thread,
};
use std::{fs, str::from_utf8, time::Duration};
use async_std::{
self,
prelude::*,
task::{self},
io::{ Read, Write }
};
pub async fn handle_connection(mut stream: impl Read + Write + Unpin) {
let mut buffer = [0; 1024];
stream.read(&mut buffer).await.unwrap();
let req: Vec<&str> = from_utf8(&buffer).unwrap().split("\r\n").filter(|item| { !item.is_empty() }).collect();
let str_list: Vec<&str> = req.get(0).unwrap().split_ascii_whitespace().collect();
let path = str_list.get(1).unwrap();
let (status_line, file_path) = match *path {
"/" => ("HTTP/1.1 200 OK", "index.html"),
"/sleep" => {
task::sleep(Duration::from_secs(5)).await;
("HTTP/1.1 200 OK", "index.html")
}
_ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
};
let res_body = fs::read_to_string(file_path).unwrap();
let res_headers = format!("Content-Type: html;\r\nContent-Length:{}", res_body.len());
let res = format!("{status_line}\r\n{res_headers}\r\n\r\n{res_body}");
stream.write(&res.as_bytes()).await.unwrap();
stream.flush().await.unwrap();
}
现在就可以来写测试元啦
既然要impl，那么我们数据的类型自然首选struct。
struct MockData {
read_data: Vec<u8>,
write_data: Vec<u8>,
}既然trait bound是Read + Write + Unpin，自然就得给我们的mock数据类型实现这仨才行。
我们先来写Unpin的，这货的最简单了，只需要咱impl即可，没有其它要求。
impl Unpin for MockData {}然后是实现Read
use super::{handle_connection};
use async_std::io::{ Read, Write };
use futures::io::Error;
use futures::task::{Context, Poll};
use std::cmp::min;
use std::pin::Pin;
struct MockData {
read_data: Vec<u8>,
write_data: Vec<u8>,
}
impl Read for MockData {
fn poll_read(
self: Pin<&mut Self>,
_: &mut Context,
buf: &mut [u8],
) -> Poll<Result<usize, Error>> {
let size: usize = min(self.read_data.len(), buf.len());
buf[..size].copy_from_slice(&self.read_data[..size]);
Poll::Ready(Ok(size))
}
}这个poll_read是必须实现的。最后返回size也符合要求。
然后我们再来实现Write的。
impl Write for MockData {
fn poll_write(
mut self: Pin<&mut Self>,
_: &mut Context,
buf: &[u8],
) -> Poll<Result<usize, Error>> {
self.write_data = Vec::from(buf);
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
}poll_write、poll_flush以及poll_close三个method也都是必须实现的。
这个没啥好说的，现在我们已经实现了一个符合要求的mock数据。
我们可以写测试函数了。
use std::fs;
#[async_std::test]
async fn test_handle_connection() {
let input_bytes = b"GET / HTTP/1.1\r\n";
let mut contents = vec![0u8; 1024];
contents[..input_bytes.len()].clone_from_slice(input_bytes);
let mut stream = MockData {
read_data: contents,
write_data: Vec::new(),
};
handle_connection(&mut stream).await;
let expected_contents = fs::read_to_string("index.html").unwrap();
let expected_headers = format!("Content-Type: html;\r\nContent-Length:{}", expected_contents.len());
let expected_response = format!("HTTP/1.1 200 OK\r\n{}\r\n\r\n{}", expected_headers,expected_contents);
assert!(stream.write_data.starts_with(expected_response.as_bytes()));
}模拟了下请求的信息，得装换成stream才行。
然后创建mock对象传给handle_connection方法，最后再response。
那么测试元也写完了。
整理下代码
#[cfg(test)]
mod tests {
use super::{handle_connection};
use async_std::io::{ Read, Write };
use futures::io::Error;
use futures::task::{Context, Poll};
use std::cmp::min;
use std::pin::Pin;
struct MockData {
read_data: Vec<u8>,
write_data: Vec<u8>,
}
impl Read for MockData {
fn poll_read(
self: Pin<&mut Self>,
_: &mut Context,
buf: &mut [u8],
) -> Poll<Result<usize, Error>> {
let size: usize = min(self.read_data.len(), buf.len());
buf[..size].copy_from_slice(&self.read_data[..size]);
Poll::Ready(Ok(size))
}
}
impl Write for MockData {
fn poll_write(
mut self: Pin<&mut Self>,
_: &mut Context,
buf: &[u8],
) -> Poll<Result<usize, Error>> {
self.write_data = Vec::from(buf);
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
fn poll_close(self: Pin<&mut Self>, _: &mut Context) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
}
impl Unpin for MockData {}
use std::fs;
#[async_std::test]
async fn test_handle_connection() {
let input_bytes = b"GET / HTTP/1.1\r\n";
let mut contents = vec![0u8; 1024];
contents[..input_bytes.len()].clone_from_slice(input_bytes);
let mut stream = MockData {
read_data: contents,
write_data: Vec::new(),
};
handle_connection(&mut stream).await;
let expected_contents = fs::read_to_string("index.html").unwrap();
let expected_headers = format!("Content-Type: html;\r\nContent-Length:{}", expected_contents.len());
let expected_response = format!("HTTP/1.1 200 OK\r\n{}\r\n\r\n{}", expected_headers,expected_contents);
assert!(stream.write_data.starts_with(expected_response.as_bytes()));
}
}我们可以用本机端口0模拟，这个端口正常来说是不允许使用的，但是可以用于测试。
不过在这我们的核心就只是这个handle_connection，所以直接测试这个测试元即可。
这代码我也放到github上，有需要的可以直接拉。
git clone https://github.com/1714080902120/rust_web_demo_async.git总结#
文档很不完善。。
参考#
- ^building-a-concurrent-web-server-with-Async-Rust https://rust-lang.github.io/async-book/09_example/00_intro.html#final-project-building-a-concurrent-web-server-with-async-rust
- ^adding-async-runtime https://rust-lang.github.io/async-book/09_example/01_running_async_code.html#adding-an-async-runtime
- ^async-std https://crates.io/crates/async-std
- ^handleing-connections-concurrency https://rust-lang.github.io/async-book/09_example/02_handling_connections_concurrently.html#handling-connections-concurrently
- ^testing-TCP-Server https://rust-lang.github.io/async-book/09_example/03_tests.html#testing-the-tcp-server
发布于 2023-02-02 14:03・IP 属地广东