所有依赖项都已升级，所有警告都已修复并升级到 2018。

目录： 概述 | 示例 | MPMC 模式 | Futures 模式 | 基准测试 | 常见问题解答

概述

多队列基于 LMAX Disruptor 的队列设计，略有改进

• 未来流/汇（实现了 futures 特性）
• 它可以动态地添加/删除生产者，并且每个 流 都可以有多个消费者
• 它在任何时候都有一对一的消费者和/或生产者时，具有快速的回退机制，并且可以在运行时检测到切换
• 它在 32 位系统上工作，没有任何性能或能力上的损失
• 在大多数情况下，可以直接在队列中写入数据，而无需复制

可以将 MultiQueue 视为一个增强版的 通道/sync_channel，具有额外的功能，即每个独立的消费者都可以接收相同的 流 数据。

选择 MultiQueue2 而不是内置通道的原因

• 支持将元素广播到多个读取器，只需在队列中单次推送即可
• 在大多数情况下，允许在队列中就地读取元素，因此您可以在不进行大量复制的情况下广播元素
• 可以作为 futures 流和汇
• 与通道不同，不会在推送/弹出时分配内存，导致具有更多可预测的延迟
• 与 sync_channel 不同，几乎是无锁的，并且在争用下表现相当不错

不选择 MultiQueue2 而选择内置通道的原因

• 真正需要无界队列，尽管您可能应该处理回压
• 需要发送者在队列满时阻塞，并且不能使用 futures API
• 不希望使用大缓冲区的内存使用量
• 需要一次性队列
• 非常频繁地添加/删除生产者/消费者

在其他情况下，MultiQueue 应该是通道的良好替代品。通常，它将像常规有界队列一样工作得很好，其性能接近手写队列的单个/多个生产者/消费者

即使不利用广播

示例

单个生产者单个流

对于队列来说，这已经是最简单的情况了。快速、一个写者、一个读者，简单易用。

extern crate multiqueue2 as multiqueue;

use std::thread;

let (send, recv) = multiqueue::mpmc_queue(10);

thread::spawn(move || {
    for val in recv {
        println!("Got {}", val);
    }
});

for i in 0..10 {
    send.try_send(i).unwrap();
}

// Drop the sender to close the queue
drop(send);

// prints
// Got 0
// Got 1
// Got 2
// etc


// some join mechanics here

单个生产者双流。

让我们将值发送到两个不同的流

extern crate multiqueue2 as multiqueue;

use std::thread;

let (send, recv) = multiqueue::broadcast_queue(4);

for i in 0..2 { // or n
    let cur_recv = recv.add_stream();
    thread::spawn(move || {
        for val in cur_recv {
            println!("Stream {} got {}", i, val);
        }
    });
}

// Take notice that I drop the reader - this removes it from
// the queue, meaning that the readers in the new threads
// won't get starved by the lack of progress from recv
recv.unsubscribe();

for i in 0..10 {
    // Don't do this busy loop in real stuff unless you're really sure
    loop {
        if send.try_send(i).is_ok() {
            break;
        }
    }
}

// Drop the sender to close the queue
drop(send);

// prints along the lines of
// Stream 0 got 0
// Stream 0 got 1
// Stream 1 got 0
// Stream 0 got 2
// Stream 1 got 1
// etc

// some join mechanics here

单个生产者广播，每个流有两个消费者

让我们将上面的内容改为每个流由两个消费者消费

extern crate multiqueue2 as multiqueue;

use std::thread;

let (send, recv) = multiqueue::broadcast_queue(4);

for i in 0..2 { // or n
    let cur_recv = recv.add_stream();
    for j in 0..2 {
        let stream_consumer = cur_recv.clone();
        thread::spawn(move || {
            for val in stream_consumer {
                println!("Stream {} consumer {} got {}", i, j, val);
            }
        });
    }
    // cur_recv is dropped here
}

// Take notice that I drop the reader - this removes it from
// the queue, meaning that the readers in the new threads
// won't get starved by the lack of progress from recv
recv.unsubscribe();

for i in 0..10 {
    // Don't do this busy loop in real stuff unless you're really sure
    loop {
        if send.try_send(i).is_ok() {
            break;
        }
    }
}
drop(send);

// prints along the lines of
// Stream 0 consumer 1 got 2
// Stream 0 consumer 0 got 0
// Stream 1 consumer 0 got 0
// Stream 0 consumer 1 got 1
// Stream 1 consumer 1 got 1
// Stream 1 consumer 0 got 2
// etc

// some join mechanics here

一些奇怪的事情

有人真正想过要使用甚至想要做更类似于的东西吗？

extern crate multiqueue2 as multiqueue;

use std::thread;

let (send, recv) = multiqueue::broadcast_queue(4);

// start like before
for i in 0..2 { // or n
    let cur_recv = recv.add_stream();
    for j in 0..2 {
        let stream_consumer = cur_recv.clone();
        thread::spawn(move || {
            for val in stream_consumer {
                println!("Stream {} consumer {} got {}", i, j, val);
            }
        });
    }
    // cur_recv is dropped here
}

// On this stream, since there's only one consumer,
// the receiver can be made into a SingleReceiver
// which can view items inline in the queue
let single_recv = recv.add_stream().into_single().unwrap();

thread::spawn(move || {
    for val in single_recv.iter_with(|item_ref| 10 * *item_ref) {
        println!("{}", val);
    }
});

// Same as above, except this time we just want to iterate until the receiver is empty
let single_recv_2 = recv.add_stream().into_single().unwrap();

thread::spawn(move || {
    for val in single_recv_2.try_iter_with(|item_ref| 10 * *item_ref) {
        println!("{}", val);
    }
});

// Take notice that I drop the reader - this removes it from
// the queue, meaning that the readers in the new threads
// won't get starved by the lack of progress from recv
recv.unsubscribe();

// Many senders to give all the receivers something
for _ in 0..3 {
    let cur_send = send.clone();
    for i in 0..10 {
        thread::spawn(loop {
            if cur_send.try_send(i).is_ok() {
                break;
            }
        });
    }
}
drop(send);

MPMC 模式