• 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!");
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    运行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);
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    BufReader 实现了BufReadtrait,提供了lines方法,通过换行符切割数据流返回一个Result迭代器。通过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();
    
    • 1
    • 2
    • 3

    重新启动,再次浏览器访问地址,可以看到空白页面,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=, initial-scale=1.0" />
        <title>Documenttitle>
      head>
      <body>
        <p>hello worldp>
      body>
    html>
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

    读取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();
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    再次运行,浏览器访问可以看到页面上已经展示信息。现在只要是所有的请求访问都会返回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();
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33

    优化一下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();
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    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);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    现在一个简易的 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" => {
                // 线程堵塞5s
                thread::sleep(Duration::from_secs(5));
                ("HTTP/1.1 200 OK", "index.html")
            }
            _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
        };
        // ...
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20

    然后我们打开两个浏览器的 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);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12

    现在可以再次尝试请求/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);
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

    实现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(&self, f: F)
        where
            F: FnOnce() + Send + 'static,
        {
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17

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

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

    然后在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>,
    }
    
    impl ThreadPool {
        /// 创建线程池
        ///
        /// 线程池中线程的数量
        ///
        /// # Panics
        ///
        /// `new`函数在size为0 时panicthread_pool
        pub fn new(size: usize) -> ThreadPool {
            assert!(size > 0);
    
            let mut threads = Vec::with_capacity(size);
    
            for _ in 0..size {
                // 创建对应数量的线程,并把它们存储到vec中
            }
    
            ThreadPool { threads }
        }
        // ...
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27

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

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

    use std::thread;
    
    pub struct ThreadPool {
        // threads: Vec>,
        workers: Vec,
    }
    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 }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38

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

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

    use std::{sync::mpsc, thread};
    
    pub struct ThreadPool {
        // threads: Vec>,
        workers: Vec,
        sender: mpsc::Sender,
    }
    
    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 }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28

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

    impl Worker {
        fn new(id: usize, receiver: mpsc::Receiver) -> Worker {
            let thread = thread::spawn(|| {
                receiver;
            });
    
            Worker { id, thread }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

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

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

    已经解决过在线程间共享状态,通过线程安全智能指针Arc>,多个线程共享所有权并允许线程修改其值。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`创建多所有者,Mutex共享数据
            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>>) -> Worker {
            // ...
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32

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

    // struct Job;
    type Job = Box;
    
    • 1
    • 2

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

    impl ThreadPool {
        pub fn execute(&self, f: F)
        where
            F: FnOnce() + Send + 'static,
        {
            let job = Box::new(f);
    
            self.sender.send(job).unwrap();
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10

    继续优化接受端执行任务的逻辑,在接收到任务后,通过lock获取互斥器来锁定资源,防止其他地方使用资源。通过unwrap处理错误时的情况,在获取了互斥器锁定了资源后,调用recv()方法接受任务Job,这会阻塞当前线程,所有如果当前线程没有任务,则会一直等待直到有用的任务。Mutex可以确保一次只有一个 Worker 线程请求任务。

    impl Worker {
        fn new(id: usize, receiver: Arc>>) -> Worker {
            let thread = thread::spawn(move || loop {
                let job = receiver.lock().unwrap().recv().unwrap();
    
                println!("开始执行任务{id}");
    
                job();
            });
    
            Worker { id, thread }
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14

    通过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();
            }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

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

    struct Worker {
        id: usize,
        // thread: thread::JoinHandle<()>,
        thread: Option>,
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5

    通过 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();
                }
            }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11

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

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

    pub struct ThreadPool {
        // threads: Vec>,
        workers: Vec,
        sender: Option>,
    }
    
    impl ThreadPool {
        pub fn new(size: usize) -> ThreadPool {
        // ...
        // ...
        ThreadPool {
            workers,
            sender: Some(sender),
        }
        }
    
        pub fn execute(&self, f: F)
        where
            F: 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) {
            // 显示的丢弃sender
            drop(self.sender.take());
    
            for worker in &mut self.workers {
                println!("stop worker {}", worker.id);
    
                if let Some(thread) = worker.thread.take() {
                    thread.join().unwrap();
                }
            }
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41

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

    impl Worker {
        fn new(id: usize, receiver: Arc>>) -> 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),
            }
        }
    }
    
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25

    现在可以正常清理、停机了,如果希望在服务停止前再处理几个请求,通过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);
            })
        }
    }
    
    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13

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

    multi-web-service-stop.png

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

  • 相关阅读:
    数据分析与处理 实验二 numpy的基本使用
    Activiti7审批流
    索引失效的几种情况
    SpringBoot单元测试
    iOS 17 适配 Xcode 15 问题
    [《与闲适共处》偶感小记]2012年8月28日
    LLM Serving有效吞吐量的最大化实现
    9/12 思维训练
    理想中的接口自动化项目
    基于Python的“书怡”在线书店系统的设计与实现毕业设计源码082332
  • 原文地址:https://blog.csdn.net/heroboyluck/article/details/134093979