Crates.io

Disruptor

这是一个用Rust编写的低延迟线程间通信库。

它深受出色的LMAX Disruptor库的启发。

入门

在你的Cargo.toml文件中添加以下内容：

disruptor = "3.2.0"

要了解如何使用该库的详细信息，请查看docs.rs/disruptor上的文档。

以下是一个最小示例，演示了单个和批量发布。注意，为了获得最佳延迟和吞吐量，应尽可能使用批量发布（参见下面的基准测试）。

use disruptor::*;

// 环形缓冲区中的事件。
struct Event {
    price: f64
}

fn main() {
    // 用于初始化环形缓冲区中事件的工厂闭包。
    let factory = || { Event { price: 0.0 }};

    // 用于处理事件的闭包。
    let processor = |e: &Event, sequence: Sequence, end_of_batch: bool| {
        // 在这里添加你的处理逻辑。
    };

    let size = 64;
    let mut producer = disruptor::build_single_producer(size, factory, BusySpin)
        .handle_events_with(processor)
        .build();

    // 通过`Producer`句柄将单个事件发布到Disruptor中。
    for i in 0..10 {
        producer.publish(|e| {
            e.price = i as f64;
        });
    }

    // 将一批事件发布到Disruptor中。
    producer.batch_publish(5, |iter| {
        for e in iter { // `iter`保证会产生5个事件。
            e.price = 42.0;
        }
    });
}// 在此处，Producer实例超出作用域，当处理器完成处理所有事件后，
 // Disruptor也会被释放。

该库还支持将线程固定在核心上，以避免上下文切换引起的延迟。一个更高级的用法，演示了这一点以及多个生产者和多个相互依赖的消费者，可能如下所示：

use disruptor::*;
use std::thread;

struct Event {
    price: f64
}

fn main() {
    let factory = || { Event { price: 0.0 }};

    // 用于处理事件的闭包。
    let h1 = |e: &Event, sequence: Sequence, end_of_batch: bool| {
        // 在这里添加处理逻辑。
    };
    let h2 = |e: &Event, sequence: Sequence, end_of_batch: bool| {
        // 在这里添加一些处理逻辑。
    };
    let h3 = |e: &Event, sequence: Sequence, end_of_batch: bool| {
        // 在这里添加更多处理逻辑。
    };

    let mut producer1 = disruptor::build_multi_producer(64, factory, BusySpin)
        // `h2`与`h1`并发处理事件。
        .pin_at_core(1).handle_events_with(h1)
        .pin_at_core(2).handle_events_with(h2)
            .and_then()
            // `h3`在`h1`和`h2`之后处理事件。
            .pin_at_core(3).handle_events_with(h3)
        .build();

    // 创建另一个生产者。
    let mut producer2 = producer1.clone();

    // 发布到Disruptor。
    thread::scope(|s| {
        s.spawn(move || {
            for i in 0..10 {
                producer1.publish(|e| {
                    e.price = i as f64;
                });
            }
        });
        s.spawn(move || {
            for i in 10..20 {
                producer2.publish(|e| {
                    e.price = i as f64;
                });
            }
        });
    });
}// 在此处，Producer实例超出作用域，当处理器完成处理所有事件后，
 // Disruptor也会被释放。

如果你需要在处理器线程中存储一些既不是Send也不是Sync的状态，例如Rc<RefCell<i32>>，那么你可以创建一个用于初始化该状态的闭包，并在构建Disruptor时将其与处理闭包一起传递。然后Disruptor将在每个事件上传递一个指向你的状态的可变引用。例如：

use std::{cell::RefCell, rc::Rc};
use disruptor::*;

struct Event {
    price: f64
}

#[derive(Default)]
struct State {
    data: Rc<RefCell<i32>>
}

fn main() {
    let factory = || { Event { price: 0.0 }};
    let initial_state = || { State::default() };

    // 用于处理事件*和*状态的闭包。
    let processor = |s: &mut State, e: &Event, _: Sequence, _: bool| {
        // 修改你的自定义状态：
        *s.data.borrow_mut() += 1;
    };

    let size = 64;
    let mut producer = disruptor::build_single_producer(size, factory, BusySpin)
        .handle_events_and_state_with(processor, initial_state)
        .build();

    for i in 0..10 {
        producer.publish(|e| {
            e.price = i as f64;
        });
    }
}

特性

单生产者单消费者（SPSC）。
单生产者多消费者（SPMC）与消费者相互依赖。
多生产者单消费者（MPSC）。
多生产者多消费者（MPMC）与消费者相互依赖。
忙等待策略。
事件批量发布。
事件批量消费。
可以为事件处理器线程设置线程亲和性。
设置每个事件处理器线程的线程名称。

设计选择

该库中的一切都是关于低延迟的，这极大地影响了该库中的所有选择。例如，你不能分配一个事件并将其移动到环形缓冲区中。相反，事件在启动时就被分配，以确保它们在内存中共同定位，从而提高缓存一致性。然而，你仍然可以在堆上分配一个结构体，并将其所有权移动到环形缓冲区上的事件字段中。只要你意识到这可能会增加延迟，因为结构体是由一个线程分配而由另一个线程释放的。因此，在分配器中会发生同步。

也没有使用动态分发 - 一切都是单态的。

正确性

这个库需要使用Unsafe来实现低延迟。虽然不能保证没有错误，但已经使用了以下方法来消除错误：

最小化Unsafe块的使用。
高测试覆盖率。
所有测试都在CI/CD中通过Miri运行。
在TLA+中进行验证（参见verification/文件夹）。

性能

SPSC和MPSC Disruptor变体已经进行了基准测试，并与Crossbeam进行了比较。请参见benches/spsc.rs和benches/mpsc.rs文件中的代码。以下是在 2016 年的 MacBook Pro（配备 2.6 GHz 四核 Intel Core i7 处理器）上运行 SPSC 基准测试的结果。在现代 Intel Xeon 处理器上，这些数字应该会更好。此外，在 Mac 上无法隔离核心并固定线程，这本可以产生更稳定的结果。这是未来的工作。

如果您有任何改进基准测试的建议，请随时提出问题。

为了提供一个相对真实的基准测试，不仅考虑了不同大小的突发，还考虑了突发之间的不同暂停时间：0 毫秒、1 毫秒和 10 毫秒。

以下延迟是每个元素的平均延迟，置信区间为 95%（标准 criterion 设置）。捕获所有延迟并计算各种百分位数（特别是最大延迟）是未来的工作。然而，我预计下面的测量结果能代表您在实际应用中可以达到的性能。

突发之间无暂停

延迟：

突发大小	Crossbeam	Disruptor	改进
1	65 ns	32 ns	51%
10	68 ns	9 ns	87%
100	29 ns	8 ns	72%

吞吐量：

突发大小	Crossbeam	Disruptor	改进
1	15.2M / s	31.7M / s	109%
10	14.5M / s	117.3M / s	709%
100	34.3M / s	119.7M / s	249%

突发之间暂停 1 毫秒

延迟：

突发大小	Crossbeam	Disruptor	改进
1	63 ns	33 ns	48%
10	67 ns	8 ns	88%
100	30 ns	9 ns	70%

吞吐量：

突发大小	Crossbeam	Disruptor	改进
1	15.9M / s	30.7M / s	93%
10	14.9M / s	117.7M / s	690%
100	33.8M / s	105.0M / s	211%

突发之间暂停 10 毫秒

延迟：

突发大小	Crossbeam	Disruptor	改进
1	51 ns	32 ns	37%
10	67 ns	9 ns	87%
100	30 ns	10 ns	67%

吞吐量：

突发大小	Crossbeam	Disruptor	改进
1	19.5M / s	31.6M / s	62%
10	14.9M / s	114.5M / s	668%
100	33.6M / s	105.0M / s	213%

结论

Disruptor 和 Crossbeam 库之间显然存在差异。然而，这并不是因为 Crossbeam 库不是一个优秀的软件。事实上它很优秀。Disruptor 通过牺牲 CPU 和内存资源来换取更低的延迟和更高的吞吐量，这就是它能够达到这些结果的原因。如基准测试中 10 和 100 个事件的突发所示，Disruptor 在发布事件批次时表现更为出色。

随着突发大小的增加，两个库的性能都有很大提升，但 Disruptor 的性能对突发之间的暂停更具韧性，这也是其设计目标之一。