advance/async/web-server

[giscus[bot]]

advance/async/web-server

https://course.rs/advance/async/web-server.html

[lqygbzy]

为什么我把依赖加上后总是build失败😥😥😥

[mxismean]

使用多线程并行处理请求:

同时发起多个 /sleep 请求， 第一个5s后返回，第二个 10s后返回，第三个15s后返回， 貌似并没有与多个线程同时处理这些请求吧？

[makisevon]

我猜你是用 Chrome 测试的？Chrome 默认使用的缓存策略限制了对同一个 URL 的并发请求，建议使用 cURL 或 Postman 等工具测试。

[mxismean]

是的，的确用的chrome， 感谢感谢🙏

[Yougigun]

記得可以把chrom devtool 打開，這個限制也會被取消

[PigsyAK]

测试结果：

1. 没有异步：同一个浏览器中
2. 没有异步：curl -Z http://127.0.0.1:7878/sleep http://127.0.0.1:7878/sleep http://127.0.0.1:7878/sleep
3. 异步：edge + chrome，edge + curl

感觉是tcp流这里阻塞了，然后浏览器不同标签页或者curl -Z的请求复用了一个tcp流。不懂，瞎猜的，希望有大佬能讲通透。

[LanYebo]

使用/sleep请求 + 正常请求同时测试可以看出来，sleep处理的同时正常请求可以返回。

[zhangdekui]

单线程异步确实神奇
listener是异步的,
handle也是异步的,
下面的sleep也是异步,
任意一个io事件发生的时候都会切换到具有异步标记的地方,
所以第一个请求异步sleep过程中, 第二个请求进来时, 异步的listener得以运行,
第二个请求运行到异步sleep时, 当第一个请求的sleep结束时, 立马又能切换过来,
看起来像是并行, 其实是并发, 切换的比较快而已.

[ranqiwei]

单线程异步模拟并行的感觉

[lgln0]

多线程+async，

spawn(handle_connection(stream));

这不是为每个async开了个线程吗，那async还有什么意义？我的理解是async类似协程，把当前线程分成一个个task，可在task间无缝切换，但task阻塞则线程阻塞。但如果每个线程里只有一个async，不等于没async吗？

[Alkaidd]

我感觉这里的例子多线程+异步确实没什么意义。但是如果你把sleep看成另一个异步任务应该就可以理解了。

[Initsnow]

这里spawn的应该是一个类似Task的东西而不是线程

[sleep-bit]

因为这里async_std的spawn函数和thread中所用的spawn函数是不一样的，前者是创建任务并将该异步任务提交到任务队列通过异步运行时来进行任务调配的，他所使用的线程数是异步运行时设定的线程池大小，后者是会创建操作系统中的线程。

[jim-yefeng]

如果任务有很多，比如 10 万个连接进来，1 个线程就不够了

[learnerLj]

tokio 版本：

use futures::stream::StreamExt;
use tokio::fs;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; // Use tokio::io::BufReader, AsyncBufReadExt, and AsyncWriteExt
use tokio::net::{TcpListener, TcpStream};
use tokio_stream::wrappers::TcpListenerStream;
#[tokio::main]
async fn main() {
    let listener = TcpListener::bind("127.0.0.1:8878").await.unwrap();
    TcpListenerStream::new(listener)
        .for_each_concurrent(None, |stream_result| async {
            match stream_result {
                Ok(stream) => {
                    handle_connection(stream).await;
                }
                Err(e) => eprintln!("Connection error: {}", e),
            }
        })
        .await;
}

async fn handle_connection(mut stream: TcpStream) {
    let buf_reader = BufReader::new(&mut stream);
    let mut lines = buf_reader.lines();
    let request_line = lines.next_line().await.unwrap().unwrap();
    let (status_line, filename) = if request_line == "GET / HTTP/1.1" {
        ("HTTP/1.1 200 OK", "hello.html")
    } else {
        ("HTTP/1.1 404 NOT FOUND", "404.html")
    };

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

cargo.toml

[package]
name = "hello"
version = "0.1.0"
edition = "2024"

[dependencies]
futures = "0.3"
tokio = { version = "1", features = ["full"] }
tokio-stream = { version = "0.1", features = ["net"] }

[skb666]

spawn(handle_connection(stream));

这一步在 tokio 中是怎么实现

[liucan233]

listener
.incoming()
.for_each_concurrent
这也太难使用了吧，有onIncoming事件实现吗

[bobobo-7]

聪明的读者不知道有没有发现，之前的例子有一个致命的缺陷：只能使用一个线程并发的处理用户请求。是的，这样也可以实现并发，一秒处理几千次请求问题不大，但是这毕竟没有利用上 CPU 的多核并行能力，无法实现性能最大化。

这里的描述似乎有问题，async-std默认的运行时应该就是多线程的，使用task::spawn并不是开启了多线程去处理并发，而是从当前的任务中分离了一个新的任务，让spawn里的内容与当前的for_each_concurrent解耦。

for_each_concurrent的第一个参数可以限制它的并发数量，把None改成Some(1)就变成串行的了，例如下面的代码中，for_each_concurrent的各个闭包变成了串行，每个闭包里又是await的，所以就是阻塞的

.for_each_concurrent(Some(1), |tcpstream| async move {
    let tcpstream = tcpstream.unwrap();
    handle_connection(tcpstream).await;
})
.await;

如果使用spawn，spawn中的代码脱离了for_each_concurrent变成了一个新的异步任务，这样就不会因为for_each_concurrent是串行的而发生阻塞了

.for_each_concurrent(Some(1), |tcpstream| async move {
    let tcpstream = tcpstream.unwrap();
    spawn(handle_connection(tcpstream));
})
.await;

[ijustlearn]

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 = MockTcpStream {
        read_data: contents,
        write_data: Vec::new(),
    };

    handle_connection(&mut stream).await;
    let mut buf = [0u8; 1024];
    stream.read(&mut buf).await.unwrap();

    let expected_contents = fs::read_to_string("hello.html").unwrap();
    let expected_response = format!("HTTP/1.1 200 OK\r\n\r\n{}", expected_contents);
    assert!(stream.write_data.starts_with(expected_response.as_bytes()));
}

最后一段测试代码中

  let mut buf = [0u8; 1024];
    stream.read(&mut buf).await.unwrap();

是不是多余了，stream.read在handle_connection函数里面已经读取过了。另外stream的read方法本来就是mock，还需要验证行为吗？