前面我们实现了一个简陋的线程池,其中线程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。