Rust并发控制之Channel

Rust 官方sync包中提供了mpsc模式的 (多生产者，单消费者：multi-producer, single-consumer) channel，可以实现基于消息并发控制，而不是依赖控制内存共享（加锁）。这正是 go 语言作者 R. Pike 所推崇的方式：

Don't communicate by sharing memory; share memory by communicating. (R. Pike)

今天就聊聊mpsc提供的sync_channel和channel。

文章目录

• 规则

• sync_channel - spsc

• sync_channel - mpsc

• channel

• 并发安全

规则

首先一般 channel 机制都保证了

• 生产者（producer/sender） 可以发送（send）消息，消费者（consumer/receiver）可以接受（recv）消息，生产和消费的顺序一致（一般都有消息队列保证顺序FIFO）
• 消费者在没有消息可接收前会阻塞等待，直到有消息或 channel 关闭
• channel 可以限制同时可处理消息上限（buffer size）
• 生产者发送的消息累积到 buffer 上限时就要阻塞到有消息被消费

从这些规则中，可以看出，channel 保证了生产总是先于消费，消息处理总是先进先出（FIFO）。

sync_channel - spsc

buffer size 最特别的情况就是 0，就是单生产者单消费者模式(mpsc)：send 后会阻塞，直到有 recv 处理，才能再 send 下一个消息。

这就能很好的实现对并发顺序的控制， 比如下边代码，用两组 channel 实现 1 和 2 的交替打印。

不同 channel 的 send 和 recv 交叉等待，保证了打印的顺序，就像这中间持有锁一样

use std::sync::mpsc::sync_channel;
use std::thread;

fn main() {
    let (sender, receiver) = sync_channel(0);
    let (sender2, receiver2) = sync_channel(0);

    let cnt = 3;
    let t1 = thread::spawn({
        move || {
            for _ in 0..cnt {
                print!("1 ");
                // t1打印完，通知t2的receiver打印
                sender.send(2).unwrap();
                // 阻塞，等待t2打印结束
                receiver2.recv().unwrap();
            }
        }
    });

    let t2 = thread::spawn({
        move || {
            for _ in 0..cnt {
                // 阻塞，等待t1 sender的已打印的消息
                receiver.recv().unwrap();
                print!("2 ");
                // t2打印完, 给t1 receiver2通知可以进行下一次打印
                sender2.send(1).unwrap();
            }
        }
    });

    t1.join().unwrap();
    t2.join().unwrap();
}

sync_channel - mpsc

buffer size 增加，就是正常mpsc摸式，可以控制同时能并发的上限（实际内部提前分配了数组来支持 buffer）。

达到上限，sender 就需要等待有 receiver 消费才能够继续发送消息。

当然没消息的话，别忘了 drop 也是可以结束 recv 一直等待消息的。

如下边代码所示：

use std::sync::mpsc::sync_channel;
use std::thread;

fn main() {
    let (sender, receiver) = sync_channel(3);

    let sender2 = sender.clone();
    let sender3 = sender.clone();
    thread::spawn(move || sender.send(1).unwrap());
    thread::spawn(move || sender2.clone().send(2).unwrap());

    drop(sender3); // 这里保证了第三个recv打印能成功
    println!("{:?}", receiver.recv().unwrap());
    println!("{:?}", receiver.recv().unwrap());
    println!("{:?}", receiver.recv());
}

channel

明白了sync_channel，channel就简单了，就是 buffer size 无限模式（实际是内部维护了一个链表自动扩容）。所有的 send 都不会阻塞，只有 recv 在没消息时需要阻塞等待 channel 中产生新的消息。

use std::sync::mpsc::{channel, sync_channel};
use std::thread;

fn main() {
    // let (sender, receiver) = sync_channel(1); // buffer为1的话，不会打印send no block
    let (sender, receiver) = channel(); // 使用channel，send不阻塞，会打印

    thread::spawn(move || {
        sender.send(1).unwrap();
        sender.send(2).unwrap();
        sender.send(3).unwrap();
        println!("send no block");
    });

    println!("{:?}", receiver.recv().unwrap());
}

如果想及时 check 是否能 recv 消息时，可以用try_recv

• TryRecvError::Empty代表目前为空，但 channel 连接还在
• TryRecvError::Disconnected则是连接已关闭，不可能再受到消息了

use std::sync::mpsc::{channel, Receiver, RecvError, TryRecvError};

fn main() {
    let (sender, receiver) = channel();
    fn try_recv_with_log(receiver: &Receiver<i32>) {
        match receiver.try_recv() {
            Ok(v) => println!("{:?}", v),
            Err(TryRecvError::Empty) => println!("error: Empty"),
            Err(TryRecvError::Disconnected) => println!("error: Disconnected"),
        }
    }
    // error: Empty
    try_recv_with_log(&receiver);
    sender.send(1).unwrap();
    receiver.recv().unwrap();
    drop(sender);
    // error: Disconnected
    try_recv_with_log(&receiver);
}

并发安全

unsafe impl<T: Send> Send for Sender<T> {}
unsafe impl<T: Send> Sync for Sender<T> {}

unsafe impl<T: Send> Send for Receiver<T> {}
impl<T> !Sync for Receiver<T> {}

最后来看看 rust 如何保证 channel 的并发安全

Sender<T>同时支持Send和Sync，其维护的消息队列可以安全的在线程间传递所有权，也可以了共享引用，即可以被多个线程同时进行 send 操作。

其中T需要实现Send, 以确保消息可以在线程间安全传递所有权，避免竞争条件或使用已释放的内存

而Receiver<T>只支持 Send，只能在线程间传递自身所有权，但不能在线程间共享引用。同时只能有一个线程拥有其所有权，进而独占的去消费Sender<T>的消息队列。

依旧是巧妙的通过Send和Sync标记 trait 保证了并发的安全，轻松实现无畏并发。