问题背景
在并发编程场景下，消息传递是一个最基本的操作，例如，在Golang中，可以简单的通过chan来传递，而在Rust中，也提供了类似的channel机制用于线程间的消息传递。
示例
use std::thread;
use std::sync::mpsc;
use std::time::Duration;
use std::cell::RefCell;
use std::rc::Rc;
pub struct Endpoint {
rx: Rc<RefCell<std::sync::mpsc::Receiver<String>>>,
}
impl Endpoint {
pub fn run(&mut self) {
for _ in 0..2 {
let j = self.rx.borrow().recv().unwrap();
println!("Got: {}", j);
}
}
}
fn th(rx: std::sync::mpsc::Receiver<String>) {
let mut ep = Endpoint {
rx: Rc::new(RefCell::new(rx)),
};
ep.run();
}
pub struct Hub {
tx: Rc<RefCell<std::sync::mpsc::Sender<String>>>,
}
fn share() -> Hub {
let (tx, rx) = mpsc::channel();
let tx1 = mpsc::Sender::clone(&tx);
thread::spawn(move || {
th(rx)
});
Hub {
tx: Rc::new(RefCell::new(tx1)),
}
}
fn main() {
let h = share();
let vals = vec![
String::from("Hello"),
String::from("Rust"),
];
for val in vals {
h.tx.borrow().send(val).unwrap();
thread::sleep(Duration::from_secs(1));
}
}
示例中使用到是mpsc结构，即多生产者-单消费者模型，如果需要mpmc的模型，就没法直接使用channel了。为了实现mpmc，可以在mpsc的基础上，对receiver加锁，多个线程抢一把锁，抢到锁的线程从channel中接收消息并处理。
use std::thread;
use std::sync::Mutex;
use std::sync::mpsc::channel;
use std::sync::Arc;
use std::rc::Rc;
use std::cell::RefCell;
use std::sync::mpsc::Sender;
use std::time::Duration;
pub trait FnBox {
fn call_box(self: Box<Self>);
}
impl<F: FnOnce()> FnBox for F {
fn call_box(self: Box<F>) {
(*self)()
}
}
pub type Task = Box<dyn FnBox + Send>;
pub struct WorkerPool {
tx: Rc<RefCell<Option<Sender<Task>>>>,
#[allow(dead_code)]
handlers: Option<Vec<thread::JoinHandle<()>>>,
}
impl WorkerPool {
pub fn new(number: usize) -> WorkerPool {
let (tx, rx) = channel::<Task>();
let mut handlers = vec![];
let rh = Arc::new(Mutex::new(rx));
for _ in 0..number {
let rh = rh.clone();
let handle = thread::spawn(move || {
while let Ok(task) = rh.lock().unwrap().recv() {
task.call_box();
}
});
handlers.push(handle);
}
WorkerPool {
tx: Rc::new(RefCell::new(Some(tx))),
handlers: Some(handlers),
}
}
pub fn dispatch(&self, t: Task) {
self.tx.borrow_mut().as_ref().unwrap().send(t);
}
}
fn main() {
let wp = WorkerPool::new(2);
let closeure = || println!("hello1");
wp.dispatch(Box::new(closeure));
let closeure = || println!("hello2");
wp.dispatch(Box::new(closeure));
let closeure = || println!("hello3");
wp.dispatch(Box::new(closeure));
let closeure = || println!("hello4");
wp.dispatch(Box::new(closeure));
thread::sleep(Duration::from_secs(1));
}
上文是基于全局锁实现的消息通信，可以改为由condition variable或者是crossbeam实现，其中crossbeam是无锁实现，但是crossbeam包含一个全局队列，并发严重时，仍然可能存在队列争抢。
为了实现无锁投递，可以为每一个worker创建一个通道，这样线程之间就不存在争抢，并且可以实现准确投递，譬如按照某种hash策略来分发任务。
更复杂的实现可以实现worker间的任务Stealing，以避免任务处理不均的情况。
网友评论