前面我们通过多线程完成了文件分片上传，但如果分片多了，会创建比较多的线程，线程多了切换就会带来比较大的开销，下面我们改成线程池的方式。rust本身没有提供线程池，我们先按照java的ThreadPoolExecutor来实现一个简单的rust线程池，大体结构如下：
1.构建一个线程池，包含一个任务队列和一个最大线程数
2.有新的任务需要执行，我们就把任务交给线程池进行调度，如果线程池的线程数量还没达到最大限制就开启新线程，达到最大限制就放入队列（队列满了等逻辑暂不考虑）
3.线程创建出来执行完自己的任务后，要维持存活继续执行还未被执行的任务，避免频繁的创建销毁（这里线程一旦创建就不在销毁，不考虑线程长时间idle的状态）
根据以上思路，代码实现如下
use std::thread;
use std::ptr::null;
use std::sync::{Mutex, Arc};
use std::collections::VecDeque;
use std::time::Duration;
use std::borrow::Borrow;
use std::rc::Rc;
use std::any::Any;
pub struct ThreadPool{
core_size:u8,
counter: u8,
queue: Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>
}
pub struct ThreadWorker{
queue: Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>,
first_task: Box<dyn FnOnce()+Send+'static>
}
impl ThreadWorker{
pub fn execute(self){
(self.first_task)();
loop {
thread::sleep(Duration::from_millis(1000));
let mut fn_queue=self.queue.lock().unwrap();
if fn_queue.len()>0 {
let task=fn_queue.pop_front();
if task.is_some() {
task.unwrap()();
}
}
}
}
}
impl ThreadPool{
pub fn new(core_size:u8)->ThreadPool{
ThreadPool{
core_size,
counter:0,
queue: Arc::new(Mutex::new(VecDeque::new()))
}
}
pub fn submit<F>(&mut self,f:F)
where F:FnOnce(),
F:Send+'static
{
if self.core_size>self.counter {
let worker=Box::new(ThreadWorker{
first_task: Box::new(f),queue:self.queue.clone()
});
self.counter=self.counter+1;
println!("init worker");
thread::spawn(move || worker.execute());
}else{
println!("add queue");
self.queue.lock().unwrap().push_back(Box::new(f));
}
}
}
测试代码：
fn main() {
let mut pool=ThreadPool::new(2);
let f=||{println!("test")};
loop {
thread::sleep(Duration::from_millis(10));
pool.submit(f);
}
}
执行代码，我们会根据打印信息发现只执行了两次创建ThreadWorker的工作，然后都是把任务添加到队列，后面不断进行执行。
具体代码逻辑就不说了，可以参考实现思路。下面介绍一些知识点
1、struct： 在之前的代码中，我们都是在文件里面直接写的方法，调用的时候都是通过模块::方法来执行的，对习惯了面向对象的java开发者来说可能会比较疑惑。其实rust也可以实现面向对象的特性。在rust中，一个rs文件是一个mod，对应的是java的package，在mod中我们可以通过struct来实现面向对象的封装特性，我们可以把struct理解成java的class，但struct里面只能写属性，不能写方法包括类似java的get和set方法，如果要赋予相关方法，可以通过impl struct实现，如：
pub struct ThreadWorker{
queue: Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>,
first_task: Box<dyn FnOnce()+Send+'static>
}
impl ThreadWorker{
pub fn execute(self){
(self.first_task)();
loop {
thread::sleep(Duration::from_millis(1000));
let mut fn_queue=self.queue.lock().unwrap();
if fn_queue.len()>0 {
let task=fn_queue.pop_front();
if task.is_some() {
task.unwrap()();
}
}
}
}
}
这样我们就创建好了一个拥有自己方法的对象，然后可以像下面这样创建对象
let worker=ThreadWorker{
first_task: XXXXX
queue: XXXXX
};
对于Javaer来说，习惯了通过构造方法来创建对象，可能还是有点不习惯，那么我们可以创建一个new方法
pub fn new(queue:Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>,
first_task:Box<dyn FnOnce()+Send+'static>)->ThreadWorker
{
ThreadWorker{
queue,first_task
}
}
这样我们就可以通过类似java的构造来创建对象了
let worker=ThreadWorker::new(
first_task: XXXXX,
queue: XXXX
);
我们都知道面向对象三个基本特性：封装、继承和多态，我们通过struct实现了封装，继承和多态我们可以通过trait来实现，代码中的dyn、FnOnce()等就是例子，后面我们写代码遇到在具体说。
2、BOX，Box是rust独享所有权的智能指针。在rust中，有些场景会要求明确类型大小，但我们又没办法确定类型大小，这时候就可以用Box进行封装，如上面代码中的
queue: Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>
这里VedDeque是rust自带的队列类型，Arc和mutex之前都用过。如果我们把Box去掉的话，编译器就会下面的错误
queue: Arc<Mutex<VecDeque<dyn FnOnce()+Send+'static>>>
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ doesn't have a size known at compile-time
在rust中，通过Box包装后会把数据保存在堆上并在栈上创建指向该数据的指针，编译器会对Box加上析构函数，用完进行释放。这点像java中的Interger等包装类型。
另外，因为Box是指针，所以当进行传递的时候，可以传递引用，不会产生数据的复制。
总体而且，Box的使用场景大体分为三类：
- 当拥有一个无法在编译时确定大小的类型，但又想要在一个要求固定尺寸的上下文环境中使用这个类型的值时。
- 当需要传递大量数据的所有权，但又不希望产生大量数据的复制行为时。
- 当希望拥有一个实现了指定trait的类型值，但又不关心具体的类型时。
3、dyn FnOnce()+Send+'static ,在线程池的代码中，我们在多个地方用到这个组合（他们都是独立的，并非一定组合在一起），主要是用来限定任务的类型，类似于java的泛型，因为要涉及的rust的trait，放到后面在说。
网友评论