Rust多线程：Worker 结构体与线程池中任务的传递机制

蓝娅萍 · 昨天 18:16

本文分享自天翼云开发者社区《Rust多线程：Worker 结构体与线程池中任务的传递机制》，作者：l****n
Rust多线程：Worker 结构体与线程池中任务的传递机制

**在实现一个多线程的 Web 服务器时，我们会遇到一个问题：如何在创建线程之后让它们在没有任务时保持等待状态，并且在任务到来时可以立即执行。这是一个典型的线程池设计问题。在 Rust 中，我们需要通过自定义设计来实现这个功能，因为标准库中的 **thread::spawn 并不直接支持这种用法。
问题描述

**Rust 的 **thread::spawn 方法会立即执行传入的闭包。如果我们想要在线程池中创建线程并让它们等待任务（即在创建时不执行任何任务），我们就需要自己设计一种机制，能够在稍后将任务传递给这些已经创建好的线程。
解决方案：引入 Worker 结构体

**为了解决这个问题，我们引入了一个 **Worker 结构体来管理线程池中的每个线程。Worker 的作用类似于一个工人，它等待任务的到来并在接收到任务时执行。
1. Worker 结构体的定义

Worker 结构体包含两个字段：

id：用于标识每个 Worker。
thread：存放线程的 JoinHandle，它是由 thread::spawn 返回的。

代码如下：

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

复制代码

2. 创建 Worker 实例

**为了让 **Worker 在没有任务时处于等待状态，我们可以在 Worker::new 函数中使用 thread::spawn 创建线程，并传入一个空的闭包：

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

复制代码

**在这里，我们创建了一个 **Worker 实例，每个 Worker 都会启动一个线程。但这个线程目前还什么都不做，因为我们传递给 spawn 的闭包是空的。
3. 将 Worker 集成到线程池中

**接下来，我们修改 **ThreadPool 的实现，使其存储 Worker 的实例而不是直接存储线程的 JoinHandle。在 ThreadPool::new 中，我们使用一个 for 循环创建多个 Worker 实例，并将它们存储在一个 Vec 中：

pub struct ThreadPool {
workers: Vec<Worker>,
}
impl ThreadPool {
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 }
}
}

复制代码

**这样，我们就为线程池创建了一个由多个 **Worker 组成的集合。每个 Worker 都有一个唯一的 ID，并且都启动了一个线程，虽然这些线程目前还没有执行任何有用的任务。
向 Worker 发送任务

现在，我们解决了创建线程并让它们等待任务的问题。接下来，我们需要设计一个机制，使得线程池能够在任务到来时将任务发送给等待中的线程。
1. 使用信道传递任务

**在 Rust 中，信道（channel）是一种非常适合在线程之间传递数据的工具。我们可以使用一个信道来传递任务。线程池会创建一个信道的发送端，每个 **Worker 会拥有信道的接收端。任务通过信道从线程池传递到 Worker，再由 Worker 中的线程执行。

use std::{sync::mpsc, thread};
pub struct ThreadPool {
workers: Vec<Worker>,
sender: mpsc::Sender<Job>,
}
struct Job;
impl ThreadPool {
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 }
}
}

复制代码

2. Worker 处理任务

**为了让 **Worker 能够处理任务，我们将信道的接收端传递给每个 Worker 的线程。线程会不断地从信道中接收任务，并执行这些任务。

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

复制代码

**不过在这段代码中，存在一个问题：信道的接收端 **receiver 被移交给了第一个 Worker，导致无法将其传递给其他 Worker。
3. 使用 Arc 和 Mutex 共享接收端

**为了解决这个问题，我们需要使用 **Arc 来共享信道的接收端，这样所有的 Worker 都可以安全地从同一个信道接收任务：

use std::{sync::{mpsc, Arc, Mutex}, thread};
type Job = Box<dyn FnOnce() + Send + 'static>;
impl ThreadPool {
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();
}
}
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 }
}
}

复制代码

**在 **Worker::new 中，线程会不断地尝试获取锁来接收任务，并在收到任务后执行。这里我们使用了 Arc 来共享接收端，使用 Mutex 来确保一次只有一个 Worker 能够接收任务。
type Job = Box 是一个动态分发的闭包（或函数），其具体实现类型在编译时不确定。Box 是一个堆分配的智能指针，用于将闭包存储在堆上。dyn FnOnce() 表示这个闭包实现了 FnOnce trait，可以被调用一次。
Send 表示这个闭包可以在线程之间安全地传递。
'static 表示闭包的生命周期是整个程序的生命周期，确保闭包在多个线程中可以安全使用。
</ul>execute 方法

**这个方法的功能是将一个新的任务（闭包）添加到线程池的任务队列中，以供线程池中的工作线程执行。下面是对 **F: FnOnce() + Send + 'static 的解释：

1. F: FnOnce() + Send + 'static
复制代码
** 是一个泛型约束，表示必须是一个实现了 FnOnce、Send和 'static的闭包类型。**
- FnOnce() 确保闭包可以被调用一次。
- Send 确保闭包可以安全地在线程之间传递。
- 'static 确保闭包的生命周期足够长，可以在整个程序运行期间有效。

**在 **execute 方法中，你将传入的闭包 f 转换成 Job 类型（即 Box

账号		自动登录	找回密码
密码			立即注册

Rust多线程：Worker 结构体与线程池中任务的传递机制

相关帖子

签约作者

Rust多线程：Worker 结构体与线程池中任务的传递机制

相关帖子

相关推荐

签约作者