前面我们实现了一个简陋的线程池，其中线程worker会轮询去线程池的队列里面获取要执行的线程，有就执行，没有就不执行。我们还可以利用rust提供的channel来改造下，改成如果有任务来了，线程池就通知worker执行的方式。我们继续保留之前的版本，在此基础上再实现一个channel的版本，可以让使用者自己选择，我们暂且称为一个主动模式，一个被动模式吧。我们可以先定义一个trait，类似java的接口，然后实现一个主动模式的worker，一个被动模式的worker，另外因为还要用到消息通知，因此还需要给线程池添加一个channel sender，实现如下：
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;
use std::sync::mpsc::{channel, Sender, Receiver};
use std::cell::RefCell;
pub enum WorkerType{
Passive,
Positive
}
pub struct ThreadPool{
core_size:u8,
counter: u8,
queue: Option<Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>>,
channel: Option<(Sender<Box<dyn FnOnce()+Send+'static>>,Arc<Mutex<Receiver<Box<dyn FnOnce()+Send+'static>>>>)>,
worker_type: WorkerType
}
trait Worker{
fn execute(self) ;
}
pub struct PassiveWorker{
first_task: Box<dyn FnOnce()+Send+'static>,
recv: Arc<Mutex<Receiver<Box<dyn FnOnce()+Send+'static>>>>
}
impl Worker for PassiveWorker{
fn execute(self){
(self.first_task)();
loop {
let task=self.recv.lock().unwrap().recv().unwrap();
task();
}
}
}
pub struct PositiveWorker{
queue: Arc<Mutex<VecDeque<Box<dyn FnOnce()+Send+'static>>>>,
first_task: Box<dyn FnOnce()+Send+'static>
}
impl Worker for PositiveWorker{
fn execute(self){
loop {
thread::sleep(Duration::from_millis(100));
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,work_type:WorkerType)->ThreadPool {
match work_type {
WorkerType::Passive=> {
let (sender, receiver) = channel();
ThreadPool {
core_size,
counter: 0,
queue: None,
channel: Option::Some((sender,Arc::new(Mutex::new(receiver)))),
worker_type: work_type
}
},
WorkerType::Positive=> {
ThreadPool{
core_size,
counter: 0,
queue: Option::Some(Arc::new(Mutex::new(VecDeque::new()))),
channel: None,
worker_type: work_type
}
}
}
}
pub fn submit<F>(&mut self,f:F)
where F:FnOnce(),
F:Send+'static
{
if self.core_size>self.counter {
println!("init worker");
match self.worker_type {
WorkerType::Passive => {
let worker =Box::new(PassiveWorker {
first_task: Box::new(f),
recv: self.channel.as_ref().unwrap().1.clone()
});
thread::spawn(move || worker.execute());
},
WorkerType::Positive => {
let worker = Box::new(PositiveWorker {
first_task: Box::new(f),
queue: self.queue.clone().unwrap()
});
thread::spawn(move || worker.execute());
}
}
self.counter=self.counter+1;
}else{
match self.worker_type {
WorkerType::Passive => {
println!("send message");
self.channel.as_ref().unwrap().0.send(Box::new(f)).unwrap();
},
WorkerType::Positive=>{
println!("add queue");
self.queue.as_ref().unwrap().lock().unwrap().push_back(Box::new(f));
}
}
}
}
}
之前我们已经接触过了struct，可以通过struct实现封装，继承和多态就通过trait来实现，类似java的interface。在java中，我们会区分extends和implement，类之间是继承，接口是实现，但在rust中，struct是不能继承的。trait的使用方式，可以参考代码中的worker和PassiveWorker、PositiveWorker。
另外，在介绍下channel和Option。先说下channel，channel是rust用来解决线程间通信提供的一种方案，可以对应成java的生产者和消费者的模式。在rust中称为mpsc即multiple producer, single consume，因为每个消息只能一个消费者消费，也就避免了多线程共享的问题。代码中我们用的是无限制缓冲的channel，发送这和接收者如果没消息发送或接收就会阻塞等待。另外rust也提供了限制消息数量的channel，可以通过sync_channel实现。
再说下Option，这个类似java8和后面版本提供的Optional，是用来解决Null空指针的问题。在rust中，是没有Null的，如果一个对象存在没有值的情况，那么就可以定义成Option，比如代码中的线程池会根据不同类型来选用不同的worker，里面的queue和channel属性会根据不同的worker来判断需不需要，所以我们定义成了Option，如果要取值的话我们可以通过match，针对不同的值做不同的处理，如：
match OptionObject{
None=>{},
Some(T)=>{}
}
另外你确认有值的话，可以暴力一点，就像上面代码里面通过unwarp() 直接获取值，不过这种方式有可能是不存在具体的值导致问题。提到unwarp，这里在再多说一点，之前的代码，我们会看到很多unwarp方法，主要是获取result对象中获取值，在rust中，通过会把返回值和错误值包装进result类型一起返回，有点像我们写api接口，通过都会封装一个result对象，里面会包括code、data和message等，在使用api的地方会根据code进行不同的处理。通常，在生产环境需要根据result进行不同的异常处理，还是不要用unwarp。
