键盘快捷键

按 ← 或 → 在章节间导航

按 S 或 / 在书中搜索

按 ? 显示此帮助

按 Esc 隐藏此帮助

从单线程服务器到多线程服务器

目前，服务器会依次处理每个请求，也就是说在第一个连接处理完成之前， 它不会处理第二个连接。如果服务器收到的请求越来越多，这种串行执行方式会 越来越低效。如果服务器收到一个需要长时间处理的请求， 后续请求将不得不等到这个长请求处理完成，即使新请求可以很快被处理。 我们需要解决这个问题，但首先让我们实际观察一下这个问题。

模拟慢请求

我们将看看处理速度慢的请求会如何影响当前服务器实现中的其他请求。 示例 21-10 实现了对 /sleep 请求的处理：通过一个模拟的慢响应， 使服务器在响应前睡眠五秒。

文件名：src/main.rs 
use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};
// --snip--

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) {
    // --snip--

    let buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    // --snip--

    let contents = fs::read_to_string(filename).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()).unwrap();
}
示例 21-10：通过睡眠五秒来模拟慢请求

现在我们有了三种情况，所以从 if 换成了 match。我们需要 显式地对 request_line 的切片进行匹配，以便与 字符串字面量值进行模式匹配；match 不会像相等方法那样 自动进行引用和解引用。

第一个分支与示例 21-9 中的 if 块相同。第二个分支 匹配对 /sleep 的请求。当收到该请求时，服务器将在 渲染成功的 HTML 页面之前休眠五秒钟。第三个分支 与示例 21-9 中的 else 块相同。

你可以看到我们的服务器有多么原始：真正的库会以 远不那么冗长的方式来处理对多个请求的识别！

使用 cargo run 启动服务器。然后，打开两个浏览器窗口：一个访问 http://127.0.0.1:7878，另一个访问 http://127.0.0.1:7878/sleep。如果你 像之前一样多次输入 / URI，你会看到它响应很快。但如果 你输入 /sleep 然后加载 /，你会看到 / 会等到 sleep 睡满整整五秒钟后才会加载。

有多种技术可以避免请求排在慢请求后面堆积， 包括像第 17 章那样使用 async；而我们将要 实现的是线程池。

使用线程池提升吞吐量

线程池（thread pool）是一组已生成并等待处理 任务的线程。当程序收到新任务时，它会将池中的 一个线程分配给该任务，该线程将处理这个任务。池中 其余的线程在第一个线程处理任务时，可用于处理 任何其他传入的任务。当第一个线程完成 其任务的处理后，它会回到空闲线程池中，准备好处理 新任务。线程池允许你并发地处理连接， 从而提升服务器的吞吐量。

我们会将池中的线程数量限制为一个较小的数字，以保护我们 免受 DoS 攻击；如果让程序为每个传入的请求都创建一个新线程， 那么向我们的服务器发起一千万个请求的人就可以通过 耗尽服务器的全部资源来造成严重破坏，使请求处理 陷入停滞。

因此，我们不会生成无限多的线程，而是让固定数量的 线程在池中等待。传入的请求会被发送到池中 进行处理。池将维护一个传入请求的队列。池中的每个 线程都会从这个队列中取出一个请求，处理该请求， 然后向队列请求另一个请求。通过这种设计，我们可以并发 处理最多 N 个请求，其中 N 是线程数。如果每个 线程都在响应一个长时间运行的请求，后续请求仍然会 在队列中堆积，但我们增加了在达到那个点之前 可以处理的长时间运行请求的数量。

这种技术只是提升 web 服务器吞吐量的众多方法之一。 你可以探索的其他选项包括 fork/join 模型、 单线程异步 I/O 模型和多线程异步 I/O 模型。如果 你对这个主题感兴趣，可以阅读更多关于其他解决方案的资料 并尝试实现它们；使用像 Rust 这样的底层语言，所有这些 选项都是可能的。

在开始实现线程池之前，让我们先谈谈使用线程池 应该是什么样子。当你尝试设计代码时，先编写客户端 接口可以帮助指导你的设计。编写代码的 API，使其 结构符合你想要的调用方式；然后，在该结构内实现 功能，而不是先实现功能再设计公有 API。

类似于我们在第 12 章项目中使用的测试驱动开发， 这里我们将使用编译器驱动开发。我们将编写调用 我们想要的功能的代码，然后查看编译器的错误，以确定 接下来应该修改什么才能让代码工作。不过在此之前， 我们将先探讨我们不会采用的技术，作为起点。

为每个请求生成一个线程

首先，让我们探讨一下如果代码为每个连接都创建一个新线程， 它可能会是什么样子。如前所述，由于可能生成无限数量线程 带来的问题，这不是我们的最终方案，但它是 先得到一个可用的多线程服务器的起点。然后，我们将添加 线程池作为改进，对比两种解决方案也会更容易。

示例 21-11 展示了对 main 所做的修改，以便在 for 循环中 生成一个新线程来处理每个流。

文件名：src/main.rs 
use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

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);
        });
    }
}

fn handle_connection(mut stream: TcpStream) {
    let buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let contents = fs::read_to_string(filename).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()).unwrap();
}
示例 21-11：为每个流生成一个新线程

正如你在第 16 章所学的，thread::spawn 将创建一个新线程，然后 在新线程中运行闭包中的代码。如果你运行这段代码并在浏览器中加载 /sleep，然后再在另外两个浏览器标签页中加载 /，你确实会看到 对 / 的请求不必等待 /sleep 完成。然而，正如 我们提到的，这最终会压垮系统，因为你会毫无限制地 创建新线程。

你可能还记得第 17 章讲过，这正是 async 和 await 真正 大放异彩的场景！在我们构建线程池时请记住这一点，并思考 如果使用 async，情况会有何不同或相同之处。

创建有限数量的线程

我们希望线程池以一种相似且熟悉的方式工作，这样从线程 切换到线程池就不需要对使用我们 API 的代码做大的改动。 示例 21-12 展示了我们想要用来替代 thread::spawn 的 ThreadPool 结构体的假想接口。

文件名：src/main.rs 
use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

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 buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let contents = fs::read_to_string(filename).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()).unwrap();
}
示例 21-12：我们理想的 ThreadPool 接口

我们使用 ThreadPool::new 来创建一个具有可配置线程数量的新线程池， 这里是四个。然后，在 for 循环中，pool.execute 具有 与 thread::spawn 类似的接口，因为它接受一个池 应该为每个流运行的闭包。我们需要实现 pool.execute，使它 接受闭包并将其交给池中的一个线程来运行。这段代码还 不能编译，但我们会尝试编译，以便编译器可以指导我们如何修复它。

使用编译器驱动开发构建 ThreadPool

对 src/main.rs 做出示例 21-12 中的修改，然后让我们利用 cargo check 的编译器错误来驱动我们的开发。这是我们得到的第一个 错误：

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
error[E0433]: cannot find type `ThreadPool` in this scope
  --> src/main.rs:11:16
   |
11 |     let pool = ThreadPool::new(4);
   |                ^^^^^^^^^^ use of undeclared type `ThreadPool`

For more information about this error, try `rustc --explain E0433`.
error: could not compile `hello` (bin "hello") due to 1 previous error

太好了！这个错误告诉我们需要一个 ThreadPool 类型或模块，所以我们现在就来 构建一个。我们的 ThreadPool 实现将独立于 web 服务器所做的工作类型。因此，让我们把 hello crate 从 二进制 crate 切换为库 crate，以容纳我们的 ThreadPool 实现。在 改为库 crate 之后，我们还可以把这个独立的线程池 库用于任何我们想用线程池来做的工作，而不仅仅是处理 web 请求。

创建一个包含以下内容的 src/lib.rs 文件，这是目前我们能拥有的最简单的 ThreadPool 结构体定义：

文件名：src/lib.rs 
pub struct ThreadPool;

然后，编辑 main.rs 文件，通过在 src/main.rs 顶部添加以下代码， 将 ThreadPool 从库 crate 引入作用域：

文件名：src/main.rs 
use hello::ThreadPool;
use std::{
    fs,
    io::{BufReader, prelude::*},
    net::{TcpListener, TcpStream},
    thread,
    time::Duration,
};

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 buf_reader = BufReader::new(&stream);
    let request_line = buf_reader.lines().next().unwrap().unwrap();

    let (status_line, filename) = match &request_line[..] {
        "GET / HTTP/1.1" => ("HTTP/1.1 200 OK", "hello.html"),
        "GET /sleep HTTP/1.1" => {
            thread::sleep(Duration::from_secs(5));
            ("HTTP/1.1 200 OK", "hello.html")
        }
        _ => ("HTTP/1.1 404 NOT FOUND", "404.html"),
    };

    let contents = fs::read_to_string(filename).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()).unwrap();
}

这段代码仍然无法工作，但让我们再检查一次以获得下一个需要 解决的错误：

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no associated function or constant named `new` found for struct `ThreadPool` in the current scope
  --> src/main.rs:12:28
   |
12 |     let pool = ThreadPool::new(4);
   |                            ^^^ associated function or constant not found in `ThreadPool`

For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error

这个错误表明接下来我们需要为 ThreadPool 创建一个名为 new 的关联函数。我们还知道 new 需要有一个能接受 4 作为参数的参数，并且应该返回一个 ThreadPool 实例。 让我们实现具有这些特征的最简单的 new 函数：

文件名：src/lib.rs 
pub struct ThreadPool;

impl ThreadPool {
    pub fn new(size: usize) -> ThreadPool {
        ThreadPool
    }
}

我们选择 usize 作为 size 参数的类型，因为我们知道线程数 为负数是没有意义的。我们还知道会把这个 4 用作线程集合的元素数量， 这正是 usize 类型的用途，正如第 3 章「整型」一节所讨论的那样。

让我们再次检查代码：

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
error[E0599]: no method named `execute` found for struct `ThreadPool` in the current scope
  --> src/main.rs:17:14
   |
17 |         pool.execute(|| {
   |         -----^^^^^^^ method not found in `ThreadPool`

For more information about this error, try `rustc --explain E0599`.
error: could not compile `hello` (bin "hello") due to 1 previous error

现在出现这个错误是因为我们在 ThreadPool 上没有 execute 方法。 回想一下「创建有限数量的线程」一节，我们决定线程池应该有一个与 thread::spawn 类似的接口。 此外，我们将实现 execute 函数，使它接受给它的闭包 并将其交给池中的一个空闲线程来运行。

我们将在 ThreadPool 上定义 execute 方法，让它接受一个闭包作为 参数。回想一下第 13 章「将捕获的值移出闭包」中的内容，我们可以用三种不同的 trait 接受闭包作为参数：Fn、FnMut 和 FnOnce。我们需要决定这里使用哪种闭包。我们知道最终 会实现类似于标准库 thread::spawn 的功能，所以我们可以看看 thread::spawn 的签名对其参数有什么约束。文档给出了如下内容：

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where
        F: FnOnce() -> T,
        F: Send + 'static,
        T: Send + 'static,

我们在这里关心的是 F 类型参数；T 类型 参数与返回值有关，我们不关心。我们可以看到 spawn 使用 FnOnce 作为 F 上的 trait 约束。这可能也是我们想要的，因为我们最终会把 execute 中得到的参数传递给 spawn。我们可以进一步确信 FnOnce 是 我们想要使用的 trait，因为运行请求的线程只会执行该 请求的闭包一次，这与 FnOnce 中的 Once 相符。

F 类型参数还具有 trait 约束 Send 和生命周期约束 'static，这些在我们的场景中很有用：我们需要 Send 来将闭包 从一个线程转移到另一个线程，需要 'static 是因为我们不知道线程执行 需要多长时间。让我们在 ThreadPool 上创建一个 execute 方法， 它将接受带有这些约束的泛型参数 F：

文件名：src/lib.rs 
pub struct ThreadPool;

impl ThreadPool {
    // --snip--
    pub fn new(size: usize) -> ThreadPool {
        ThreadPool
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

我们仍然在 FnOnce 后面使用 ()，因为这个 FnOnce 表示一个 不接收任何参数并返回单元类型 () 的闭包。就像函数 定义一样，返回类型可以从签名中省略，但即使没有 参数，我们仍然需要括号。

同样，这是 execute 方法最简单的实现：它什么 都不做，但我们只是想让代码编译通过。让我们再检查一下：

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.24s

编译通过了！但请注意，如果你尝试 cargo run 并在浏览器中发出请求， 你会在浏览器中看到本章开头看到的错误。我们的库还没有真正 调用传递给 execute 的闭包！

注意：关于像 Haskell 和 Rust 这样拥有严格编译器的语言， 你可能会听到一种说法：「如果代码编译通过了，它就能正常工作。」但这种说法并非 普遍成立。我们的项目编译通过了，但它什么都没做！如果 我们正在构建一个真实、完整的项目，这将是开始 编写单元测试以检查代码编译并且具有我们想要的行为的好时机。

思考一下：如果我们要执行的是 future 而不是闭包， 这里会有什么不同？

在 new 中校验线程数量

我们还没有对 new 和 execute 的参数做任何事情。让我们 用期望的行为来实现这些函数的函数体。首先，让我们想想 new。之前我们为 size 参数选择了无符号类型，因为线程数为负数的线程池没有意义。 然而，零个线程的线程池也没有意义，但零是一个完全 合法的 usize。我们会添加代码来检查 size 大于零， 然后再返回 ThreadPool 实例，并使用 assert! 宏让程序在 收到零时 panic，如示例 21-13 所示。

文件名：src/lib.rs 
pub struct ThreadPool;

impl ThreadPool {
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        ThreadPool
    }

    // --snip--
    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}
示例 21-13：实现 ThreadPool::new，当 size 为零时 panic

我们还为 ThreadPool 添加了带文档注释的文档。 请注意，我们遵循了良好的文档实践，添加了一个小节来 指出我们的函数可能 panic 的情况，正如第 14 章所讨论的。试着运行 cargo doc --open 并点击 ThreadPool 结构体， 看看为 new 生成的文档是什么样子的！

除了像这里这样添加 assert! 宏之外，我们还可以将 new 改为 build 并返回一个 Result，就像我们在 I/O 项目示例 12-9 中对 Config::build 所做的那样。但我们在这里决定，尝试创建 一个没有任何线程的线程池应该是一个不可恢复的错误。如果你 有兴致，可以尝试编写一个名为 build 的函数，其签名如下， 以便与 new 函数进行比较：

pub fn build(size: usize) -> Result<ThreadPool, PoolCreationError> {

创建存储线程的空间

既然我们有办法知道要在池中存储的线程数量是合法的， 我们就可以创建这些线程，并在返回结构体之前将它们存储在 ThreadPool 结构体 中。但是如何“存储”一个线程呢？让我们再来 看看 thread::spawn 的签名：

pub fn spawn<F, T>(f: F) -> JoinHandle<T>
    where
        F: FnOnce() -> T,
        F: Send + 'static,
        T: Send + 'static,

spawn 函数返回一个 JoinHandle<T>，其中 T 是闭包返回的类型。让我们也尝试使用 JoinHandle 看看会发生什么。在我们的 例子中，传递给线程池的闭包将处理连接 而不返回任何内容，所以 T 将是单元类型 ()。

示例 21-14 中的代码可以编译，但还没有创建任何线程。 我们修改了 ThreadPool 的定义，使其持有一个 thread::JoinHandle<()> 实例的向量，用 size 作为容量初始化该向量，设置了一个 for 循环来运行一些创建线程的代码，并 返回包含这些线程的 ThreadPool 实例。

文件名：src/lib.rs 
use std::thread;

pub struct ThreadPool {
    threads: Vec<thread::JoinHandle<()>>,
}

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let mut threads = Vec::with_capacity(size);

        for _ in 0..size {
            // create some threads and store them in the vector
        }

        ThreadPool { threads }
    }
    // --snip--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}
示例 21-14：为 ThreadPool 创建一个 vector 来持有线程

我们在库 crate 中将 std::thread 引入作用域，因为我们在 ThreadPool 中使用 thread::JoinHandle 作为 vector 中元素的类型。

一旦收到合法的 size，我们的 ThreadPool 就会创建一个可以 容纳 size 个元素的新 vector。with_capacity 函数执行与 Vec::new 相同的任务，但有一个重要区别：它会预先在 vector 中分配空间。 因为我们知道需要在 vector 中存储 size 个元素，所以预先 进行这种分配比使用 Vec::new 稍微高效一些，后者会在插入元素时 自行扩容。

当你再次运行 cargo check 时，应该会成功。

将代码从 ThreadPool 发送到线程

我们在示例 21-14 的 for 循环中留下了一条关于创建 线程的注释。在这里，我们来看看实际上是如何创建线程的。标准 库提供了 thread::spawn 作为创建线程的方式，而 thread::spawn 期望在创建线程时立即获得线程应该运行的代码。 然而，在我们的场景中，我们希望创建线程并让它们等待 我们稍后发送的代码。标准库对线程的实现没有提供任何 实现这一点的方法；我们必须手动实现。

我们将通过在 ThreadPool 和线程之间引入一个新的数据结构 来实现这个行为，该数据结构将管理这种新行为。我们把这个 数据结构称为 Worker，这是池化实现中的常用术语。 Worker 会取出需要运行的代码并在其线程中运行这些代码。

想象一下在餐厅厨房工作的人们：工人们等待顾客的订单进来， 然后他们负责接收这些订单并完成它们。

我们将不再在线程池中存储 JoinHandle<()> 实例的 vector， 而是存储 Worker 结构体的实例。每个 Worker 将存储一个 JoinHandle<()> 实例。然后，我们将在 Worker 上实现一个方法， 它会接受要运行的代码闭包，并将其发送到已经运行的线程中 执行。我们还会给每个 Worker 一个 id，以便在日志记录或 调试时区分池中不同的 Worker 实例。

下面是创建 ThreadPool 时将发生的新流程。在以这种方式 设置好 Worker 之后，我们将实现把闭包发送到线程的代码：

  1. 定义一个持有 id 和 JoinHandle<()> 的 Worker 结构体。
  2. 修改 ThreadPool 使其持有 Worker 实例的 vector。
  3. 定义一个 Worker::new 函数，它接受一个 id 编号并返回一个 持有该 id 和一个用空闭包生成的线程的 Worker 实例。
  4. 在 ThreadPool::new 中，使用 for 循环计数器生成一个 id， 用该 id 创建一个新的 Worker，并将该 Worker 存储在 vector 中。

如果你想挑战一下，可以在查看示例 21-15 中的代码之前 试着自己实现这些修改。

准备好了吗？示例 21-15 给出了进行上述修改的一种方式。

文件名：src/lib.rs 
use std::thread;

pub struct ThreadPool {
    workers: Vec<Worker>,
}

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        let mut workers = Vec::with_capacity(size);

        for id in 0..size {
            workers.push(Worker::new(id));
        }

        ThreadPool { workers }
    }
    // --snip--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

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

impl Worker {
    fn new(id: usize) -> Worker {
        let thread = thread::spawn(|| {});

        Worker { id, thread }
    }
}
示例 21-15：修改 ThreadPool 使其持有 Worker 实例而不是直接持有线程

我们把 ThreadPool 上的字段名从 threads 改为了 workers， 因为它现在持有的是 Worker 实例而不是 JoinHandle<()> 实例。我们使用 for 循环中的计数器作为 Worker::new 的参数，并将每个新的 Worker 存储在名为 workers 的向量中。

外部代码（比如我们 src/main.rs 中的服务器）不需要知道 在 ThreadPool 内部使用 Worker 结构体的实现细节， 因此我们将 Worker 结构体及其 new 函数设为私有。 Worker::new 函数使用我们给它的 id，并存储一个通过使用空闭包创建新线程而生成的 JoinHandle<()> 实例。

注意：如果操作系统因系统资源不足而无法创建线程，thread::spawn 将会 panic。这将导致我们的 整个服务器 panic，即使其中一些线程的创建可能 成功。为了简单起见，这种行为是可以接受的，但在生产级的 线程池实现中，你可能希望改用 std::thread::Builder及其返回 Result 的 spawn方法。

这段代码可以编译，并且会存储我们在 ThreadPool::new 参数中指定数量的 Worker 实例。但我们仍然没有处理 在 execute 中获取的闭包。接下来让我们看看如何做到这一点。

通过通道向线程发送请求

我们要解决的下一个问题是，传给 thread::spawn 的闭包完全 什么也不做。目前，我们在 execute 方法中获取想要执行的闭包。但我们需要在创建 ThreadPool 期间 创建每个 Worker 时，给 thread::spawn 一个要运行的闭包。

我们希望刚创建的 Worker 结构体能从 ThreadPool 持有的队列中获取要运行的代码，并将该代码发送到它的线程中运行。

我们在第 16 章学到的通道——一种在两个线程之间进行通信的简单方式—— 非常适合这个用例。我们将使用通道来充当 任务队列，execute 会将任务从 ThreadPool 发送到 Worker 实例，后者再将任务发送到它的线程。计划如下：

  1. ThreadPool 将创建一个通道并持有发送端。
  2. 每个 Worker 将持有接收端。
  3. 我们将创建一个新的 Job 结构体，用于持有我们想要通过通道发送 的闭包。
  4. execute 方法会通过 发送端发送它想要执行的任务。
  5. 在它的线程中，Worker 将循环遍历它的接收端，并执行 收到的任何任务的闭包。

让我们首先在 ThreadPool::new 中创建一个通道，并将发送端持有 在 ThreadPool 实例中，如示例 21-16 所示。Job 结构体 目前不持有任何内容，但它将是我们通过通道发送的 元素的类型。

文件名：src/lib.rs 
use std::{sync::mpsc, thread};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

struct Job;

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }
    // --snip--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

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

impl Worker {
    fn new(id: usize) -> Worker {
        let thread = thread::spawn(|| {});

        Worker { id, thread }
    }
}
示例 21-16：修改 ThreadPool 使其存储用于传输 Job 实例的通道的发送端

在 ThreadPool::new 中，我们创建了新的通道，并让线程池持有 发送端。这样可以成功编译。

让我们尝试在线程池创建通道时，将通道的接收端传递给每个 Worker。我们知道希望在 Worker 实例所创建的线程中使用接收端，因此我们将在 闭包中引用 receiver 参数。示例 21-17 中的代码还不能完全编译通过。

文件名：src/lib.rs 
use std::{sync::mpsc, thread};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

struct Job;

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }
    // --snip--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

// --snip--


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

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

        Worker { id, thread }
    }
}
示例 21-17：将接收端传递给每个 Worker

我们做了一些微小而直接的改动：将接收端传入 Worker::new，然后在闭包内部使用它。

当我们尝试检查这段代码时，会得到如下错误：

$ cargo check
    Checking hello v0.1.0 (file:///projects/hello)
error[E0382]: use of moved value: `receiver`
  --> src/lib.rs:26:42
   |
21 |         let (sender, receiver) = mpsc::channel();
   |                      -------- move occurs because `receiver` has type `std::sync::mpsc::Receiver<Job>`, which does not implement the `Copy` trait
...
25 |         for id in 0..size {
   |         ----------------- inside of this loop
26 |             workers.push(Worker::new(id, receiver));
   |                                          ^^^^^^^^ value moved here, in previous iteration of loop
   |
note: consider changing this parameter type in method `new` to borrow instead if owning the value isn't necessary
  --> src/lib.rs:47:33
   |
47 |     fn new(id: usize, receiver: mpsc::Receiver<Job>) -> Worker {
   |        --- in this method       ^^^^^^^^^^^^^^^^^^^ this parameter takes ownership of the value

For more information about this error, try `rustc --explain E0382`.
error: could not compile `hello` (lib) due to 1 previous error

这段代码试图将 receiver 传递给多个 Worker 实例。这 行不通，回想一下第 16 章的内容：Rust 提供的通道实现是多 生产者、单消费者的。这意味着我们不能 仅仅克隆通道的消费端来修复这段代码。我们也不想 将一条消息多次发送给多个消费者；我们想要一个 消息列表和多个 Worker 实例，使得每条消息只被 处理一次。

此外，从通道队列中取出一个任务涉及对 receiver 的可变操作，因此这些线程需要一种安全的方式来共享和修改 receiver； 否则，我们可能会遇到竞态条件（如第 16 章所述）。

回想一下第 16 章讨论的线程安全智能指针：要在 多个线程之间共享所有权并允许线程修改值，我们 需要使用 Arc<Mutex<T>>。Arc 类型将让多个 Worker 实例 拥有接收端的所有权，而 Mutex 将确保同一时间只有一个 Worker 能从 接收端获取任务。示例 21-18 展示了我们需要做的修改。

文件名：src/lib.rs 
use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};
// --snip--

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

struct Job;

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }

    // --snip--

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
    }
}

// --snip--

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

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

        Worker { id, thread }
    }
}
示例 21-18：使用 Arc 和 Mutex 在 Worker 实例之间共享接收端

在 ThreadPool::new 中，我们将接收端放入一个 Arc 和一个 Mutex 中。对于每个 新的 Worker，我们克隆 Arc 来增加引用计数，以便这些 Worker 实例可以共享接收端的所有权。

有了这些修改，代码可以编译了！我们离目标越来越近了！

实现 execute 方法

让我们最终在 ThreadPool 上实现 execute 方法。我们还将把 Job 从一个结构体改为一个 trait 对象的类型别名，该 trait 对象持有 execute 接收的闭包类型。正如第 20 章“类型同义词与类型 别名”一节所讨论的，类型别名 允许我们将长类型缩短以便于使用。请看示例 21-19。

文件名：src/lib.rs 
use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

// --snip--

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

impl ThreadPool {
    // --snip--
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);

        self.sender.send(job).unwrap();
    }
}

// --snip--

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

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

        Worker { id, thread }
    }
}
示例 21-19：为 Box 创建一个 Job 类型别名来持有每个闭包，然后将任务通过通道发送出去

在使用 execute 中获取的闭包创建了一个新的 Job 实例之后，我们 将该任务发送到通道的发送端。我们在 send 上调用 unwrap 以防发送失败。例如，如果我们 停止了所有线程的执行，就意味着接收端已经停止 接收新消息，这种情况就可能发生。目前，我们还不能停止线程的 执行：只要线程池存在，我们的线程就会一直执行。我们使用 unwrap 的原因是我们知道这种失败不会发生，但 编译器并不知道这一点。

但我们还没有完全完成！在 Worker 中，传递给 thread::spawn 的闭包仍然只是引用了通道的接收端。 相反，我们需要闭包永远循环，向通道的 接收端请求任务，并在得到任务时运行它。让我们对 Worker::new 做出示例 21-20 所示的修改。

文件名：src/lib.rs 
use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

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

impl ThreadPool {
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);

        self.sender.send(job).unwrap();
    }
}

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

// --snip--

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 {id} got a job; executing.");

                job();
            }
        });

        Worker { id, thread }
    }
}
示例 21-20：在 Worker 实例的线程中接收并执行任务

在这里，我们首先对 receiver 调用 lock 来获取互斥锁，然后 调用 unwrap 在任何错误发生时 panic。如果互斥锁处于 中毒（poisoned）状态，获取锁可能会失败，这种情况可能发生在某个其他线程在 持有锁时 panic 而不是释放锁。在这种情况下，调用 unwrap 让这个线程 panic 是正确的做法。你可以随意将 这个 unwrap 改为带有对你有意义的错误信息的 expect。

如果获得了互斥锁，我们就调用 recv 从通道接收一个 Job。最后的 unwrap 也跳过了这里可能发生的任何错误，如果 持有发送端的线程已经关闭就可能出现错误，这类似于 接收端关闭时 send 方法返回 Err 的情况。

对 recv 的调用会阻塞，所以如果还没有任务，当前线程将 一直等待，直到有任务可用。Mutex<T> 确保同一时间只有一个 Worker 线程在尝试请求任务。

我们的线程池现在处于可工作状态了！运行 cargo run 并发起一些 请求：

$ cargo run
   Compiling hello v0.1.0 (file:///projects/hello)
warning: field `workers` is never read
 --> src/lib.rs:7:5
  |
6 | pub struct ThreadPool {
  |            ---------- field in this struct
7 |     workers: Vec<Worker>,
  |     ^^^^^^^
  |
  = note: `#[warn(dead_code)]` on by default

warning: fields `id` and `thread` are never read
  --> src/lib.rs:48:5
   |
47 | struct Worker {
   |        ------ fields in this struct
48 |     id: usize,
   |     ^^
49 |     thread: thread::JoinHandle<()>,
   |     ^^^^^^

warning: `hello` (lib) generated 2 warnings
    Finished `dev` profile [unoptimized + debuginfo] target(s) in 4.91s
     Running `target/debug/hello`
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.
Worker 1 got a job; executing.
Worker 3 got a job; executing.
Worker 0 got a job; executing.
Worker 2 got a job; executing.

成功了！我们现在拥有了一个可以异步执行连接的线程池。 创建的线程永远不会超过四个，因此如果服务器收到大量请求，我们的系统也不会 过载。如果我们向 /sleep 发起请求，服务器将能够通过让另一个 线程运行其他请求来为它们提供服务。

注意：如果你同时在多个浏览器窗口中打开 /sleep，它们 可能会以五秒的间隔逐个加载。某些 Web 浏览器出于缓存原因会顺序执行 同一请求的多个实例。这一 限制并非由我们的 Web 服务器造成。

现在适合暂停一下，思考如果我们使用 future 而不是闭包来完成 工作，示例 21-18、21-19 和 21-20 中的代码会有哪些不同。哪些类型会改变？方法签名会 有何不同（如果会的话）？代码的哪些部分会保持不变？

在学习了第 17 章和第 19 章中的 while let 循环之后，你 可能会好奇为什么我们没有将 Worker 线程代码写成示例 21-21 所示的样子。

文件名：src/lib.rs 
use std::{
    sync::{Arc, Mutex, mpsc},
    thread,
};

pub struct ThreadPool {
    workers: Vec<Worker>,
    sender: mpsc::Sender<Job>,
}

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

impl ThreadPool {
    /// Create a new ThreadPool.
    ///
    /// The size is the number of threads in the pool.
    ///
    /// # Panics
    ///
    /// The `new` function will panic if the size is zero.
    pub fn new(size: usize) -> ThreadPool {
        assert!(size > 0);

        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 }
    }

    pub fn execute<F>(&self, f: F)
    where
        F: FnOnce() + Send + 'static,
    {
        let job = Box::new(f);

        self.sender.send(job).unwrap();
    }
}

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

impl Worker {
    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
        let thread = thread::spawn(move || {
            while let Ok(job) = receiver.lock().unwrap().recv() {
                println!("Worker {id} got a job; executing.");

                job();
            }
        });

        Worker { id, thread }
    }
}
示例 21-21：使用 while let 的 Worker::new 替代实现

这段代码可以编译和运行，但并不能产生期望的线程 行为：一个慢请求仍然会导致其他请求等待被 处理。原因有些微妙：Mutex 结构体没有公开的 unlock 方法，因为锁的所有权基于 lock 方法返回的 LockResult<MutexGuard<T>> 中 MutexGuard<T> 的生命周期。在编译时，借用检查器可以借此强制执行 这样一条规则：除非我们持有锁，否则无法访问由 Mutex 守护的 资源。然而，如果我们不注意 MutexGuard<T> 的生命周期，这种实现也可能导致锁被持有的 时间比预期更长。

示例 21-20 中使用 let job = receiver.lock().unwrap().recv().unwrap(); 的代码之所以能正常工作，是因为使用 let 时，等号 右侧表达式中使用的任何 临时值都会在 let 语句结束时立即被丢弃。然而，while let（以及 if let 和 match）直到 关联代码块结束时才会丢弃临时值。在示例 21-21 中，锁在 调用 job() 期间一直被持有，这意味着其他 Worker 实例无法接收任务。