7.2 线程安全队列¶
1.disruptor
Disruptor 通过模块化组件设计实现低延迟并发通信，各组件功能与协作逻辑明确，具体如下：
| 组件名称 | 核心功能 | 关键特性 |
|---|---|---|
| 生产者（Producer） | 生成并发布事件到环形缓冲区 | 单生产者场景无竞争，多生产者通过 CAS 操作抢占缓冲区槽位，避免锁开销 |
| 环形缓冲区（Ring Buffer） | 存储事件的核心数据结构 | 预分配固定大小，无运行时内存分配 / 回收，支持指针数组或结构体数组存储事件容器 |
| 序列器（Sequencer） | 管理生产者 / 消费者的序列号同步 | 作为并发控制核心，通过序列号协调数据读写顺序，确保生产者与消费者操作的一致性 |
| 事件处理器（Event Processor，即消费者） | 从环形缓冲区读取并处理事件 | 支持多消费者并行或按依赖链处理（如流水线多阶段任务），通过自身序列号跟踪处理进度，无需锁同步 |
| 序列屏障（Sequence Barrier） | 控制消费者对事件的读取时机 | 消费者通过屏障检查序列号是否可用，结合等待策略（如忙等、yield、阻塞）决定等待方式，确保安全读取 |
| 等待策略（Wait Strategy） | 定义消费者无事件可处理时的行为 | 平衡延迟与 CPU 占用：忙等策略低延迟但高 CPU，阻塞策略低 CPU 但高延迟，yield 策略为中间方案 |
#ifndef DISRUPTOR_HPP
#define DISRUPTOR_HPP
#include <algorithm>
#include <array>
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <limits>
#include <memory>
#include <new> // for std::align
#include <stdexcept>
#include <thread>
#include <vector>
// --- 平台相关的性能优化 ---
#if defined(_MSC_VER) && (defined(_M_IX86) || defined(_M_X64))
#include <intrin.h>
#define HAS_MM_PAUSE 1
#elif (defined(__GNUC__) || defined(__clang__)) && \
(defined(__i386__) || defined(__x86_64__))
#include <immintrin.h>
#define HAS_MM_PAUSE 1
#else
#define HAS_MM_PAUSE 0
#endif
#ifdef __cpp_lib_hardware_interference_size
static constexpr size_t CACHE_LINE_SIZE =
std::hardware_destructive_interference_size;
#else
static constexpr size_t CACHE_LINE_SIZE = 64;
#endif
template <typename T>
struct alignas(CACHE_LINE_SIZE) AlignedAtomic {
std::atomic<T> value;
AlignedAtomic(const AlignedAtomic&) = delete;
AlignedAtomic& operator=(const AlignedAtomic&) = delete;
AlignedAtomic(AlignedAtomic&&) = delete;
AlignedAtomic& operator=(AlignedAtomic&&) = delete;
AlignedAtomic(T initial_value = T()) : value(initial_value) {}
};
using Sequence = AlignedAtomic<long long>;
namespace disruptor {
enum class ProducerType { SINGLE, MULTI };
template <typename T>
class ExceptionHandler {
public:
virtual ~ExceptionHandler() = default;
virtual void handleEventException(const std::exception& ex,
long long sequence, T* event) = 0;
virtual void handleOnStartException(const std::exception& ex) = 0;
virtual void handleOnShutdownException(const std::exception& ex) = 0;
};
template <typename T>
class DefaultExceptionHandler : public ExceptionHandler<T> {
public:
void handleEventException(const std::exception& ex, long long sequence,
T* event) override {
std::cerr << "Disruptor: Exception on seq " << sequence << ": " << ex.what()
<< std::endl;
}
void handleOnStartException(const std::exception& ex) override {
std::cerr << "Disruptor: Exception on start: " << ex.what() << std::endl;
}
void handleOnShutdownException(const std::exception& ex) override {
std::cerr << "Disruptor: Exception on shutdown: " << ex.what() << std::endl;
}
};
template <typename T>
class RingBuffer {
private:
const size_t capacity_;
const size_t indexMask_;
std::unique_ptr<char[]> buffer_;
T* entries_;
static bool isPowerOfTwo(size_t n) { return (n > 0) && ((n & (n - 1)) == 0); }
public:
explicit RingBuffer(size_t capacity)
: capacity_(capacity), indexMask_(capacity - 1) {
if (!isPowerOfTwo(capacity)) {
throw std::invalid_argument(
"RingBuffer capacity must be a power of two.");
}
size_t buffer_size = capacity * sizeof(T) + alignof(T) - 1;
buffer_ = std::make_unique<char[]>(buffer_size);
void* ptr = buffer_.get();
size_t space = buffer_size;
entries_ = static_cast<T*>(
std::align(alignof(T), capacity * sizeof(T), ptr, space));
for (size_t i = 0; i < capacity; ++i) {
new (&entries_[i]) T();
}
}
~RingBuffer() {
for (size_t i = 0; i < capacity_; ++i) {
entries_[i].~T();
}
}
size_t getCapacity() const { return capacity_; }
T& get(long long sequence) { return entries_[sequence & indexMask_]; }
};
class WaitStrategy {
public:
virtual ~WaitStrategy() = default;
virtual long long waitFor(long long sequence, Sequence& cursor,
const std::vector<Sequence*>& dependentSequences,
const std::atomic<bool>& running) = 0;
virtual void signalAllWhenBlocking() = 0;
};
inline long long getGatingSequence(
const Sequence& cursor, const std::vector<Sequence*>& dependentSequences) {
if (dependentSequences.empty()) {
return cursor.value.load(std::memory_order_acquire);
}
long long minVal = std::numeric_limits<long long>::max();
for (const auto* dep : dependentSequences) {
minVal = std::min(minVal, dep->value.load(std::memory_order_acquire));
}
return minVal;
}
class BlockingWaitStrategy : public WaitStrategy {
private:
std::mutex mutex_;
std::condition_variable condition_;
public:
long long waitFor(long long sequence, Sequence& cursor,
const std::vector<Sequence*>& dependentSequences,
const std::atomic<bool>& running) override {
std::unique_lock<std::mutex> lock(mutex_);
condition_.wait(lock, [&] {
return !running.load(std::memory_order_relaxed) ||
getGatingSequence(cursor, dependentSequences) >= sequence;
});
return std::min(cursor.value.load(std::memory_order_acquire),
getGatingSequence(cursor, dependentSequences));
}
void signalAllWhenBlocking() override { condition_.notify_all(); }
};
class YieldingWaitStrategy : public WaitStrategy {
public:
long long waitFor(long long sequence, Sequence& cursor,
const std::vector<Sequence*>& dependentSequences,
const std::atomic<bool>& running) override {
int counter = 100;
while (getGatingSequence(cursor, dependentSequences) < sequence &&
running.load(std::memory_order_relaxed)) {
if (counter > 0) {
--counter;
} else {
std::this_thread::yield();
}
}
return std::min(cursor.value.load(std::memory_order_acquire),
getGatingSequence(cursor, dependentSequences));
}
void signalAllWhenBlocking() override {}
};
class BusySpinWaitStrategy : public WaitStrategy {
public:
long long waitFor(long long sequence, Sequence& cursor,
const std::vector<Sequence*>& dependentSequences,
const std::atomic<bool>& running) override {
while (getGatingSequence(cursor, dependentSequences) < sequence &&
running.load(std::memory_order_relaxed)) {
#if HAS_MM_PAUSE
_mm_pause();
#endif
}
return std::min(cursor.value.load(std::memory_order_acquire),
getGatingSequence(cursor, dependentSequences));
}
void signalAllWhenBlocking() override {}
};
class SleepingWaitStrategy : public WaitStrategy {
private:
const int spin_retries_ = 100;
const int yield_retries_ = 100;
const std::chrono::nanoseconds sleepTimeNs_;
public:
explicit SleepingWaitStrategy(long long sleepTimeNs = 100000)
: sleepTimeNs_(sleepTimeNs) {}
long long waitFor(long long sequence, Sequence& cursor,
const std::vector<Sequence*>& dependentSequences,
const std::atomic<bool>& running) override {
int counter = spin_retries_ + yield_retries_;
while (getGatingSequence(cursor, dependentSequences) < sequence &&
running.load(std::memory_order_relaxed)) {
if (counter > yield_retries_) {
--counter;
} else if (counter > 0) {
--counter;
std::this_thread::yield();
} else {
std::this_thread::sleep_for(sleepTimeNs_);
}
}
return std::min(cursor.value.load(std::memory_order_acquire),
getGatingSequence(cursor, dependentSequences));
}
void signalAllWhenBlocking() override {}
};
class ISequencer {
public:
virtual ~ISequencer() = default;
virtual long long next(int n = 1) = 0;
virtual void publish(long long sequence) = 0;
virtual void publish(long long lo, long long hi) = 0;
virtual long long getCursor() const = 0;
virtual Sequence& getCursorSequence() = 0;
virtual void addGatingSequences(const std::vector<Sequence*>& sequences) = 0;
virtual void removeGatingSequence(Sequence* sequence) = 0;
};
class MultiProducerSequencer : public ISequencer {
private:
WaitStrategy& waitStrategy_;
Sequence cursor_;
std::vector<Sequence*> gatingSequences_;
const size_t capacity_;
Sequence claimCursor_;
std::unique_ptr<AlignedAtomic<long long>[]> availableBuffer_;
public:
MultiProducerSequencer(size_t bufferSize, WaitStrategy& waitStrategy)
: waitStrategy_(waitStrategy),
cursor_(-1),
capacity_(bufferSize),
claimCursor_(-1),
availableBuffer_(
std::make_unique<AlignedAtomic<long long>[]>(bufferSize)) {
for (size_t i = 0; i < capacity_; ++i) {
availableBuffer_[i].value.store(-1);
}
}
void addGatingSequences(const std::vector<Sequence*>& sequences) override {
gatingSequences_.insert(gatingSequences_.end(), sequences.begin(),
sequences.end());
}
void removeGatingSequence(Sequence* sequence) override {
auto it =
std::remove(gatingSequences_.begin(), gatingSequences_.end(), sequence);
gatingSequences_.erase(it, gatingSequences_.end());
}
long long getCursor() const override {
return cursor_.value.load(std::memory_order_acquire);
}
Sequence& getCursorSequence() override { return cursor_; }
long long next(int n = 1) override {
if (n <= 0) throw std::invalid_argument("n must be positive.");
long long currentClaim =
claimCursor_.value.fetch_add(n, std::memory_order_acq_rel);
long long lo = currentClaim + 1;
long long hi = currentClaim + n;
long long wrapPoint = hi - capacity_;
while (wrapPoint > getMinimumGatingSequenceFromConsumers()) {
std::this_thread::yield();
}
return lo;
}
void publish(long long sequence) override { publish(sequence, sequence); }
void publish(long long lo, long long hi) override {
for (long long l = lo; l <= hi; ++l) {
setAvailable(l);
}
long long currentCursorValue =
cursor_.value.load(std::memory_order_relaxed);
while (true) {
long long nextSequence = currentCursorValue + 1;
if (nextSequence > hi) {
waitStrategy_.signalAllWhenBlocking();
break;
}
if (isAvailable(nextSequence)) {
if (!cursor_.value.compare_exchange_weak(
currentCursorValue, nextSequence, std::memory_order_release,
std::memory_order_relaxed)) {
}
} else {
waitStrategy_.signalAllWhenBlocking();
break;
}
}
}
private:
void setAvailable(long long sequence) {
availableBuffer_[sequence & (capacity_ - 1)].value.store(
sequence / capacity_, std::memory_order_release);
}
bool isAvailable(long long sequence) {
return availableBuffer_[sequence & (capacity_ - 1)].value.load(
std::memory_order_acquire) == (sequence / capacity_);
}
long long getMinimumGatingSequenceFromConsumers() {
if (gatingSequences_.empty()) {
return std::numeric_limits<long long>::max();
}
long long minVal = std::numeric_limits<long long>::max();
for (Sequence* gate : gatingSequences_) {
minVal = std::min(minVal, gate->value.load(std::memory_order_acquire));
}
return minVal;
}
};
class SingleProducerSequencer : public ISequencer {
private:
WaitStrategy& waitStrategy_;
Sequence cursor_;
Sequence cachedGatingSequence_;
std::vector<Sequence*> gatingSequences_;
const size_t capacity_;
long long nextValue_ = -1;
public:
SingleProducerSequencer(size_t bufferSize, WaitStrategy& waitStrategy)
: waitStrategy_(waitStrategy),
cursor_(-1),
cachedGatingSequence_(-1),
capacity_(bufferSize) {}
void addGatingSequences(const std::vector<Sequence*>& sequences) override {
gatingSequences_.insert(gatingSequences_.end(), sequences.begin(),
sequences.end());
}
void removeGatingSequence(Sequence* sequence) override {
auto it =
std::remove(gatingSequences_.begin(), gatingSequences_.end(), sequence);
gatingSequences_.erase(it, gatingSequences_.end());
}
long long getCursor() const override {
return cursor_.value.load(std::memory_order_acquire);
}
Sequence& getCursorSequence() override { return cursor_; }
long long next(int n = 1) override {
if (n <= 0) throw std::invalid_argument("n must be positive.");
long long lo = nextValue_ + 1;
long long hi = nextValue_ + n;
long long wrapPoint = hi - capacity_;
long long cachedGate =
cachedGatingSequence_.value.load(std::memory_order_acquire);
if (wrapPoint > cachedGate ||
cachedGate > cursor_.value.load(std::memory_order_relaxed)) {
long long minSequence;
while (wrapPoint >
(minSequence = getMinimumGatingSequenceFromConsumers())) {
std::this_thread::yield();
}
cachedGatingSequence_.value.store(minSequence, std::memory_order_release);
}
nextValue_ = hi;
return lo;
}
void publish(long long sequence) override {
cursor_.value.store(sequence, std::memory_order_release);
waitStrategy_.signalAllWhenBlocking();
}
void publish(long long lo, long long hi) override { publish(hi); }
private:
long long getMinimumGatingSequenceFromConsumers() {
if (gatingSequences_.empty()) {
return std::numeric_limits<long long>::max();
}
long long minVal = std::numeric_limits<long long>::max();
for (Sequence* gate : gatingSequences_) {
minVal = std::min(minVal, gate->value.load(std::memory_order_acquire));
}
return minVal;
}
};
template <typename T>
class EventHandler {
public:
virtual ~EventHandler() = default;
virtual void onEvent(T& event, long long sequence, bool endOfBatch) = 0;
virtual void onStart() {}
virtual void onShutdown() {}
};
template <typename T>
class EventProcessor {
private:
std::unique_ptr<Sequence> sequence_;
RingBuffer<T>& ringBuffer_;
ISequencer& sequencer_;
WaitStrategy& waitStrategy_;
EventHandler<T>& eventHandler_;
ExceptionHandler<T>& exceptionHandler_;
std::vector<Sequence*> dependentSequences_;
std::atomic<bool> running_{false};
std::thread thread_;
public:
EventProcessor(RingBuffer<T>& ringBuffer, ISequencer& sequencer,
WaitStrategy& waitStrategy, EventHandler<T>& eventHandler,
ExceptionHandler<T>& exceptionHandler)
: sequence_(new Sequence(-1)),
ringBuffer_(ringBuffer),
sequencer_(sequencer),
waitStrategy_(waitStrategy),
eventHandler_(eventHandler),
exceptionHandler_(exceptionHandler) {}
void addDependentSequences(const std::vector<Sequence*>& sequences) {
dependentSequences_.insert(dependentSequences_.end(), sequences.begin(),
sequences.end());
}
void run() {
running_.store(true);
try {
eventHandler_.onStart();
} catch (const std::exception& ex) {
exceptionHandler_.handleOnStartException(ex);
running_.store(false);
return;
}
long long nextSequence = sequence_->value.load() + 1;
while (running_.load()) {
try {
const long long availableSequence =
waitStrategy_.waitFor(nextSequence, sequencer_.getCursorSequence(),
dependentSequences_, running_);
if (!running_.load()) break;
if (nextSequence <= availableSequence) {
while (nextSequence <= availableSequence) {
eventHandler_.onEvent(ringBuffer_.get(nextSequence), nextSequence,
nextSequence == availableSequence);
nextSequence++;
}
sequence_->value.store(availableSequence, std::memory_order_release);
// NOTE:
// 消费者也是生产者。当它更新自己的进度时，必须通知可能正在等待它的下游消费者。
waitStrategy_.signalAllWhenBlocking();
} else {
// 如果没有事件可处理，可以短暂让出CPU，避免在某些等待策略中空转
std::this_thread::sleep_for(std::chrono::nanoseconds(1));
}
} catch (const std::exception& ex) {
if (running_.load()) {
exceptionHandler_.handleEventException(
ex, nextSequence, &ringBuffer_.get(nextSequence));
sequence_->value.store(nextSequence);
nextSequence++;
}
}
}
try {
eventHandler_.onShutdown();
} catch (const std::exception& ex) {
exceptionHandler_.handleOnShutdownException(ex);
}
}
void start() { thread_ = std::thread(&EventProcessor::run, this); }
void halt() {
running_.store(false);
waitStrategy_.signalAllWhenBlocking();
if (thread_.joinable()) {
thread_.join();
}
}
Sequence& getSequence() const { return *sequence_; }
};
template <typename T>
class Disruptor {
private:
RingBuffer<T> ringBuffer_;
std::unique_ptr<WaitStrategy> waitStrategy_;
std::unique_ptr<ISequencer> sequencer_;
std::vector<std::unique_ptr<EventProcessor<T>>> eventProcessors_;
std::unique_ptr<ExceptionHandler<T>> exceptionHandler_;
public:
Disruptor(
size_t capacity, ProducerType producerType = ProducerType::SINGLE,
WaitStrategy* strategy = new BlockingWaitStrategy(),
ExceptionHandler<T>* exceptionHandler = new DefaultExceptionHandler<T>())
: ringBuffer_(capacity),
waitStrategy_(strategy),
exceptionHandler_(exceptionHandler) {
if (producerType == ProducerType::SINGLE) {
sequencer_ =
std::make_unique<SingleProducerSequencer>(capacity, *waitStrategy_);
} else {
sequencer_ =
std::make_unique<MultiProducerSequencer>(capacity, *waitStrategy_);
}
}
EventProcessor<T>* handleEventsWith(
EventHandler<T>& handler,
const std::vector<EventProcessor<T>*>& dependencies = {}) {
for (auto dep_proc : dependencies) {
sequencer_->removeGatingSequence(&dep_proc->getSequence());
}
auto processor = std::make_unique<EventProcessor<T>>(
ringBuffer_, *sequencer_, *waitStrategy_, handler, *exceptionHandler_);
std::vector<Sequence*> depSequences;
for (auto dep : dependencies) {
depSequences.push_back(&dep->getSequence());
}
processor->addDependentSequences(depSequences);
sequencer_->addGatingSequences({&processor->getSequence()});
auto* processorPtr = processor.get();
eventProcessors_.push_back(std::move(processor));
return processorPtr;
}
void start() {
for (auto& processor : eventProcessors_) {
processor->start();
}
}
void shutdown() {
for (auto it = eventProcessors_.rbegin(); it != eventProcessors_.rend();
++it) {
(*it)->halt();
}
}
ISequencer& getSequencer() { return *sequencer_; }
RingBuffer<T>& getRingBuffer() { return ringBuffer_; }
void publishEvent(
std::function<void(T& event, long long sequence)> translator) {
const long long sequence = sequencer_->next();
try {
translator(ringBuffer_.get(sequence), sequence);
} catch (...) {
sequencer_->publish(sequence);
throw;
}
sequencer_->publish(sequence);
}
void publishEvents(
std::function<void(T& event, long long sequence)> translator,
int batchSize) {
if (batchSize <= 0) {
return;
}
const long long lo = sequencer_->next(batchSize);
const long long hi = lo + (batchSize - 1);
try {
for (long long sequence = lo; sequence <= hi; ++sequence) {
translator(ringBuffer_.get(sequence), sequence);
}
} catch (...) {
sequencer_->publish(lo, hi);
throw;
}
sequencer_->publish(lo, hi);
}
};
} // namespace disruptor
#endif // DISRUPTOR_HPP
#include <iomanip>
#include <iostream>
#include <numeric>
#include <string>
#include <vector>
#include "disruptor.hpp"
// ------------------------------------------------------------------
// 1. 测试用的事件和事件处理器定义
// ------------------------------------------------------------------
// 定义一个事件结构体
struct TestEvent {
long long value;
std::string source;
// 默认构造函数是必须的，因为RingBuffer会提前创建对象
TestEvent() : value(0), source("default") {}
};
// 打印处理器，用于最终消费和验证
class PrintEventHandler : public disruptor::EventHandler<TestEvent> {
private:
int handlerId_;
bool silent_; // 添加一个成员变量来控制是否打印
public:
// 构造函数：添加一个带默认值的布尔参数
explicit PrintEventHandler(int id, bool silent = false)
: handlerId_(id), silent_(silent) {}
void onEvent(TestEvent& event, long long sequence, bool endOfBatch) override {
// 使用 silent_ 成员变量来决定是否执行打印操作
if (!silent_) {
// 保持输出整洁，只在特定条件下打印
if (sequence < 100 || sequence % 1000000 == 0) {
std::cout << "[Handler " << handlerId_ << "] Seq: " << sequence
<< ", Value: " << event.value << ", Source: " << event.source
<< (endOfBatch ? " (EndOfBatch)\n" : "\n");
}
}
}
};
// 模拟写入日志的处理器
class JournalingEventHandler : public disruptor::EventHandler<TestEvent> {
public:
void onEvent(TestEvent& event, long long sequence, bool endOfBatch) override {
event.source += " -> Journaled";
}
};
// 模拟数据复制的处理器
class ReplicatingEventHandler : public disruptor::EventHandler<TestEvent> {
public:
void onEvent(TestEvent& event, long long sequence, bool endOfBatch) override {
event.source += " -> Replicated";
}
};
// 一个简单的函数，用于等待消费者处理完所有事件
void waitForProcessing(disruptor::EventProcessor<TestEvent>& processor,
long long expected_sequence) {
while (processor.getSequence().value.load(std::memory_order_acquire) <
expected_sequence) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
}
// ------------------------------------------------------------------
// 2. 各种并发场景的演示
// ------------------------------------------------------------------
void run_spsc_scenario() {
std::cout
<< "\n--- SCENARIO 1: Single Producer, Single Consumer (SPSC) ---\n";
disruptor::Disruptor<TestEvent> disruptor(1024,
disruptor::ProducerType::SINGLE);
PrintEventHandler handler(1); // 此处调用是正确的，因为 silent 默认为 false
auto* processor = disruptor.handleEventsWith(handler);
disruptor.start();
const int event_count = 10;
std::cout << "Publisher: Sending " << event_count << " events...\n";
for (int i = 0; i < event_count; ++i) {
disruptor.publishEvent([&](TestEvent& event, long long seq) {
event.value = i;
event.source = "Producer";
});
}
waitForProcessing(*processor, event_count - 1);
disruptor.shutdown();
std::cout << "SPSC scenario finished.\n";
}
void run_spmc_scenario() {
std::cout << "\n--- SCENARIO 2: Single Producer, Multiple Consumers (SPMC "
"Broadcast) ---\n";
disruptor::Disruptor<TestEvent> disruptor(1024,
disruptor::ProducerType::SINGLE);
PrintEventHandler handler1(1);
PrintEventHandler handler2(2);
auto* processor1 = disruptor.handleEventsWith(handler1);
auto* processor2 = disruptor.handleEventsWith(handler2);
disruptor.start();
const int event_count = 10;
std::cout << "Publisher: Sending " << event_count
<< " events to 2 consumers...\n";
for (int i = 0; i < event_count; ++i) {
disruptor.publishEvent([&](TestEvent& event, long long seq) {
event.value = i;
event.source = "Producer";
});
}
waitForProcessing(*processor1, event_count - 1);
waitForProcessing(*processor2, event_count - 1);
disruptor.shutdown();
std::cout << "SPMC scenario finished.\n";
}
void run_mpmc_scenario() {
std::cout << "\n--- SCENARIO 3: Multiple Producers, Multiple Consumers "
"(MPMC) ---\n";
disruptor::Disruptor<TestEvent> disruptor(1024,
disruptor::ProducerType::MULTI);
PrintEventHandler handler1(1);
PrintEventHandler handler2(2);
auto* processor1 = disruptor.handleEventsWith(handler1);
auto* processor2 = disruptor.handleEventsWith(handler2);
disruptor.start();
const int events_per_producer = 5;
auto producer_task = [&](int producer_id) {
for (int i = 0; i < events_per_producer; ++i) {
disruptor.publishEvent([&](TestEvent& event, long long seq) {
event.value = (producer_id * 100) + i;
event.source = "Producer " + std::to_string(producer_id);
});
}
};
std::cout << "Two producers starting, each sending " << events_per_producer
<< " events...\n";
std::thread p1(producer_task, 1);
std::thread p2(producer_task, 2);
p1.join();
p2.join();
waitForProcessing(*processor1, (events_per_producer * 2) - 1);
waitForProcessing(*processor2, (events_per_producer * 2) - 1);
disruptor.shutdown();
std::cout << "MPMC scenario finished.\n";
}
void run_dependency_graph_scenario() {
std::cout
<< "\n--- SCENARIO 4: Dependency Graph (Diamond: P -> (A, B) -> C) ---\n";
disruptor::Disruptor<TestEvent> disruptor(1024);
JournalingEventHandler handlerA;
ReplicatingEventHandler handlerB;
PrintEventHandler handlerC(3);
auto* procA = disruptor.handleEventsWith(handlerA);
auto* procB = disruptor.handleEventsWith(handlerB);
auto* procC = disruptor.handleEventsWith(handlerC, {procA, procB});
disruptor.start();
const int event_count = 5;
std::cout << "Publisher: Sending " << event_count
<< " events into a diamond dependency graph...\n";
for (int i = 0; i < event_count; ++i) {
disruptor.publishEvent([&](TestEvent& event, long long seq) {
event.value = i;
event.source = "Producer";
});
}
waitForProcessing(*procC, event_count - 1);
disruptor.shutdown();
std::cout << "Dependency graph scenario finished. Note the 'source' field in "
"the output.\n";
}
// ------------------------------------------------------------------
// 3. 等待策略性能对比
// ------------------------------------------------------------------
void run_wait_strategy_comparison() {
std::cout << "\n--- PERFORMANCE TEST: Comparing Wait Strategies ---"
<< std::endl;
std::cout << "Each test will run a SPSC scenario with 5,000,000 events.\n";
const long long event_count = 5000000;
const size_t buffer_size = 2048;
auto run_test = [&](const std::string& name,
disruptor::WaitStrategy* strategy) {
std::cout << std::left << std::setw(25) << name << ": " << std::flush;
disruptor::Disruptor<TestEvent> disruptor(
buffer_size, disruptor::ProducerType::SINGLE, strategy);
PrintEventHandler silent_handler(1, true);
auto* processor = disruptor.handleEventsWith(silent_handler);
disruptor.start();
auto start_time = std::chrono::high_resolution_clock::now();
// 不能一次性发布远超缓冲区大小的事件。改为分批发布。
const int batch_size = 1024; // 设置一个合理的批次大小
for (long long i = 0; i < event_count; i += batch_size) {
long long count_in_batch =
std::min((long long)batch_size, event_count - i);
disruptor.publishEvents(
[&](TestEvent& event, long long seq) { event.value = seq; },
count_in_batch);
}
waitForProcessing(*processor, event_count - 1);
auto end_time = std::chrono::high_resolution_clock::now();
std::chrono::duration<double, std::milli> duration = end_time - start_time;
disruptor.shutdown();
std::cout << std::fixed << std::setprecision(2) << duration.count()
<< " ms\n";
};
run_test("BusySpinWaitStrategy", new disruptor::BusySpinWaitStrategy());
run_test("YieldingWaitStrategy", new disruptor::YieldingWaitStrategy());
run_test("SleepingWaitStrategy", new disruptor::SleepingWaitStrategy());
run_test("BlockingWaitStrategy", new disruptor::BlockingWaitStrategy());
}
// ------------------------------------------------------------------
// 主函数
// ------------------------------------------------------------------
int main() {
std::cout << "===============================================\n";
std::cout << " Disruptor C++ Final Implementation Test\n";
std::cout << "===============================================\n";
run_spsc_scenario();
run_spmc_scenario();
run_mpmc_scenario();
run_dependency_graph_scenario();
run_wait_strategy_comparison();
return 0;
}
zhixuan@gpu04:~/disruptor$ ./disruptor_test
===============================================
Disruptor C++ Final Implementation Test
===============================================
--- SCENARIO 1: Single Producer, Single Consumer (SPSC) ---
Publisher: Sending 10 events...
[Handler 1] Seq: 0, Value: 0, Source: Producer
[Handler 1] Seq: 1, Value: 1, Source: Producer
[Handler 1] Seq: 2, Value: 2, Source: Producer
[Handler 1] Seq: 3, Value: 3, Source: Producer
[Handler 1] Seq: 4, Value: 4, Source: Producer
[Handler 1] Seq: 5, Value: 5, Source: Producer
[Handler 1] Seq: 6, Value: 6, Source: Producer
[Handler 1] Seq: 7, Value: 7, Source: Producer
[Handler 1] Seq: 8, Value: 8, Source: Producer
[Handler 1] Seq: 9, Value: 9, Source: Producer (EndOfBatch)
SPSC scenario finished.
--- SCENARIO 2: Single Producer, Multiple Consumers (SPMC Broadcast) ---
Publisher: Sending 10 events to 2 consumers...
[Handler 1] Seq: 0, Value: 0, Source: Producer
[Handler 1] Seq: 1, Value: 1, Source: Producer
[Handler 1] Seq: 2, Value: 2, Source: Producer
[Handler 1] Seq: 3, Value: 3, Source: Producer
[Handler 2] Seq: 0, Value: 0, Source: Producer
[Handler 2] Seq: 1, Value: 1, Source: Producer
[Handler 2] Seq: 2, Value: 2, Source: Producer
[Handler 2] Seq: 3, Value: 3, Source: Producer
[Handler 2] Seq: 4, Value: 4, Source: Producer
[Handler 2] Seq: 5, Value: 5, Source: Producer
[Handler 2] Seq: 6, Value: 6, Source: Producer
[Handler [Handler 1] Seq: 42, Value: ] Seq: 4, Source: Producer
[Handler 1] Seq: 5, Value: 5, Source: Producer
[Handler 1] Seq: 6, Value: 6, Source: Producer
7[Handler 1] Seq: 7, Value: , Value: 77, Source: Producer
[Handler 2] Seq: 8, Value: 8, Source: Producer
[Handler 2] Seq: 9, Value: 9, Source: Producer (EndOfBatch)
, Source: Producer
[Handler 1] Seq: 8, Value: 8, Source: Producer
[Handler 1] Seq: 9, Value: 9, Source: Producer (EndOfBatch)
SPMC scenario finished.
--- SCENARIO 3: Multiple Producers, Multiple Consumers (MPMC) ---
Two producers starting, each sending 5 events...
[Handler 1] Seq: 0, Value: 100, Source: Producer 1
[Handler 1] Seq: 1, Value: 101, Source: Producer 1
[Handler 1] Seq: 2, Value: 102, Source: Producer 1
[Handler 1] Seq: 3, Value: 103, Source: Producer 1
[Handler 1] Seq: 4, Value: 104, Source: Producer 1 (EndOfBatch)
[Handler 1] Seq: 5, Value: 200, Source: Producer 2
[Handler 1] Seq: 6, Value: 201, Source: Producer 2
[Handler 1] Seq: 7, Value: 202, Source: Producer 2
[Handler 1] Seq: 8, Value: 203, Source: Producer 2
[Handler 1] Seq: 9, Value: 204, Source: Producer 2 (EndOfBatch)
[Handler 2] Seq: 0, Value: 100, Source: Producer 1
[Handler 2] Seq: 1, Value: 101, Source: Producer 1
[Handler 2] Seq: 2, Value: 102, Source: Producer 1
[Handler 2] Seq: 3, Value: 103, Source: Producer 1
[Handler 2] Seq: 4, Value: 104, Source: Producer 1 (EndOfBatch)
[Handler 2] Seq: 5, Value: 200, Source: Producer 2
[Handler 2] Seq: 6, Value: 201, Source: Producer 2
[Handler 2] Seq: 7, Value: 202, Source: Producer 2
[Handler 2] Seq: 8, Value: 203, Source: Producer 2
[Handler 2] Seq: 9, Value: 204, Source: Producer 2 (EndOfBatch)
MPMC scenario finished.
--- SCENARIO 4: Dependency Graph (Diamond: P -> (A, B) -> C) ---
Publisher: Sending 5 events into a diamond dependency graph...
[Handler 3] Seq: 0, Value: 0, Source: Producer -> Replicated
[Handler 3] Seq: 1, Value: 1, Source: Producer -> Journaled -> Replicated
[Handler 3] Seq: 2, Value: 2, Source: Producer -> Journaled -> Replicated
[Handler 3] Seq: 3, Value: 3, Source: Producer -> Journaled -> Replicated
[Handler 3] Seq: 4, Value: 4, Source: Producer -> Journaled -> Replicated (EndOfBatch)
Dependency graph scenario finished. Note the 'source' field in the output.
--- PERFORMANCE TEST: Comparing Wait Strategies ---
Each test will run a SPSC scenario with 5,000,000 events.
BusySpinWaitStrategy : 31.57 ms
YieldingWaitStrategy : 29.59 ms
SleepingWaitStrategy : 29.27 ms
BlockingWaitStrategy : 29.47 ms
2.与dpdk的rte_ring相同机制的无锁队列
#include <immintrin.h> // for _mm_pause()
#include <atomic>
#include <cstdint>
#include <iostream>
#include <memory>
#include <vector>
template <typename T>
class RingBuffer {
public:
enum class ProducerType { SINGLE, MULTI };
enum class ConsumerType { SINGLE, MULTI };
RingBuffer(size_t capacity, ProducerType prod_type = ProducerType::SINGLE,
ConsumerType cons_type = ConsumerType::SINGLE)
: prod_type_(prod_type), cons_type_(cons_type) {
// Ensure capacity is power of 2
size_t actual_capacity = 1;
while (actual_capacity < capacity) {
actual_capacity <<= 1;
}
capacity_ = actual_capacity;
mask_ = capacity_ - 1;
buffer_.resize(capacity_);
prod_head_.store(0, std::memory_order_relaxed);
prod_tail_.store(0, std::memory_order_relaxed);
cons_head_.store(0, std::memory_order_relaxed);
cons_tail_.store(0, std::memory_order_relaxed);
}
bool enqueue(T item) { return enqueue(&item, 1) == 1; }
size_t enqueue(T* items, size_t count) {
if (count == 0 || count > capacity_) return 0;
uint32_t prod_head, prod_next, free_entries;
const uint32_t mask = mask_;
if (prod_type_ == ProducerType::SINGLE) {
prod_head = prod_head_.load(std::memory_order_relaxed);
do {
uint32_t cons_tail = cons_tail_.load(std::memory_order_acquire);
free_entries = (capacity_ + cons_tail - prod_head) & mask_;
if (free_entries < count) return 0;
prod_next = (prod_head + count) & mask_;
} while (false); // Single producer doesn't need retry
prod_head_.store(prod_next, std::memory_order_release);
for (size_t i = 0; i < count; i++) {
uint32_t idx = (prod_head + i) & mask;
buffer_[idx] = items[i];
}
prod_tail_.store(prod_next, std::memory_order_release);
return count;
} else {
uint32_t initial_prod_head;
do {
prod_head = prod_head_.load(std::memory_order_relaxed);
initial_prod_head = prod_head;
uint32_t cons_tail = cons_tail_.load(std::memory_order_acquire);
free_entries = (capacity_ + cons_tail - prod_head) & mask_;
if (free_entries < count) return 0;
prod_next = (prod_head + count) & mask_;
} while (!prod_head_.compare_exchange_weak(prod_head, prod_next,
std::memory_order_release,
std::memory_order_relaxed));
for (size_t i = 0; i < count; i++) {
uint32_t idx = (prod_head + i) & mask;
buffer_[idx] = items[i];
}
// Wait for predecessor producers to update tail
while (prod_tail_.load(std::memory_order_relaxed) != initial_prod_head) {
_mm_pause(); // Reduce CPU consumption during busy wait
}
prod_tail_.store(prod_next, std::memory_order_release);
return count;
}
}
bool dequeue(T& item) { return dequeue(&item, 1) == 1; }
size_t dequeue(T* items, size_t count) {
if (count == 0 || count > capacity_) return 0;
uint32_t cons_head, cons_next, entries;
const uint32_t mask = mask_;
if (cons_type_ == ConsumerType::SINGLE) {
cons_head = cons_head_.load(std::memory_order_relaxed);
do {
uint32_t prod_tail = prod_tail_.load(std::memory_order_acquire);
entries = (prod_tail - cons_head) & mask_;
if (entries < count) return 0;
cons_next = (cons_head + count) & mask_;
} while (false); // Single consumer doesn't need retry
cons_head_.store(cons_next, std::memory_order_release);
for (size_t i = 0; i < count; i++) {
uint32_t idx = (cons_head + i) & mask;
items[i] = std::move(buffer_[idx]);
}
cons_tail_.store(cons_next, std::memory_order_release);
return count;
} else {
uint32_t initial_cons_head;
// 1. Acquire consumption range via CAS on cons_head
do {
cons_head = cons_head_.load(std::memory_order_relaxed);
initial_cons_head = cons_head;
uint32_t prod_tail = prod_tail_.load(std::memory_order_acquire);
entries = (prod_tail - cons_head) & mask_;
if (entries < count) return 0;
cons_next = (cons_head + count) & mask_;
} while (!cons_head_.compare_exchange_weak(cons_head, cons_next,
std::memory_order_release,
std::memory_order_relaxed));
// 2. Copy data from buffer
for (size_t i = 0; i < count; i++) {
uint32_t idx = (initial_cons_head + i) & mask;
items[i] = std::move(buffer_[idx]);
}
// 3. Lazy update cons_tail - only update if we're next in line
uint32_t current_cons_tail;
do {
current_cons_tail = cons_tail_.load(std::memory_order_relaxed);
if (current_cons_tail != initial_cons_head) {
_mm_pause(); // Reduce CPU consumption during busy wait
continue;
}
} while (!cons_tail_.compare_exchange_weak(current_cons_tail, cons_next,
std::memory_order_release,
std::memory_order_relaxed));
return count;
}
}
size_t count() const {
uint32_t prod_tail = prod_tail_.load(std::memory_order_acquire);
uint32_t cons_head = cons_head_.load(std::memory_order_acquire);
return (prod_tail - cons_head) & mask_;
}
bool empty() const { return count() == 0; }
bool full() const { return count() == capacity_; }
size_t capacity() const { return capacity_; }
private:
ProducerType prod_type_;
ConsumerType cons_type_;
size_t capacity_;
size_t mask_;
std::vector<T> buffer_;
alignas(64) std::atomic<uint32_t> prod_head_;
alignas(64) std::atomic<uint32_t> prod_tail_;
alignas(64) std::atomic<uint32_t> cons_head_;
alignas(64) std::atomic<uint32_t> cons_tail_;
};