Disruptor для Java: полное руководство с примерами

1. Что это и зачем нужна

Disruptor — это высокопроизводительная библиотека для межпотоковой коммуникации (IPC) на Java, разработанная компанией LMAX. Она предназначена для передачи данных между потоками с минимальными задержками и максимальной пропускной способностью. В отличие от традиционных очередей (например, LinkedBlockingQueue), Disruptor использует кольцевой буфер (ring buffer) фиксированного размера, что позволяет избежать накладных расходов на выделение памяти, синхронизацию и сборку мусора.

Основная проблема, которую решает Disruptor — это «соревнование» потоков за ресурсы. В традиционных очередях несколько продюсеров и консумеров конкурируют за блокировки, что приводит к простоям и снижению производительности. Disruptor использует механизм «барьеров последовательности» (sequence barriers) и оптимистичную блокировку (CAS — Compare-And-Swap), что позволяет достичь задержек в микросекундах и пропускной способности в миллионы сообщений в секунду.

Библиотека идеально подходит для систем, где критична скорость обработки событий: биржевые платформы, игровые серверы, системы обработки логов, высоконагруженные веб-приложения. Disruptor является основой для многих современных фреймворков, включая Apache Storm и некоторые реализации Event Sourcing.

2. Установка

Disruptor распространяется через Maven Central. Для использования в проекте добавьте зависимость в ваш pom.xml (Maven):

<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>4.0.0</version>
</dependency>

Для Gradle (build.gradle):

implementation 'com.lmax:disruptor:4.0.0'

Если вы используете другой менеджер зависимостей, скачайте JAR с Maven Central.

3. Быстрый старт — минимальный рабочий пример

В этом примере мы создадим простой Disruptor, который принимает строковые события и выводит их в консоль.

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;

// 1. Определяем класс события
public class StringEvent {
    private String value;

    public String getValue() { return value; }
    public void setValue(String value) { this.value = value; }
}

// 2. Фабрика событий (создает пустые события для переиспользования)
public class StringEventFactory implements EventFactory<StringEvent> {
    @Override
    public StringEvent newInstance() {
        return new StringEvent();
    }
}

// 3. Обработчик событий (Consumer)
public class StringEventHandler implements EventHandler<StringEvent> {
    @Override
    public void onEvent(StringEvent event, long sequence, boolean endOfBatch) {
        System.out.println("Получено: " + event.getValue());
    }
}

// 4. Главный класс
public class QuickStart {
    public static void main(String[] args) throws InterruptedException {
        // Размер кольцевого буфера (должен быть степенью двойки)
        int bufferSize = 1024;

        // Создаем Disruptor
        Disruptor<StringEvent> disruptor = new Disruptor<>(
            new StringEventFactory(),
            bufferSize,
            DaemonThreadFactory.INSTANCE
        );

        // Подключаем обработчик
        disruptor.handleEventsWith(new StringEventHandler());

        // Запускаем Disruptor
        disruptor.start();

        // Получаем ссылку на RingBuffer
        RingBuffer<StringEvent> ringBuffer = disruptor.getRingBuffer();

        // Публикуем события
        for (int i = 0; i < 10; i++) {
            long sequence = ringBuffer.next(); // Захватываем слот
            try {
                StringEvent event = ringBuffer.get(sequence);
                event.setValue("Событие #" + i);
            } finally {
                ringBuffer.publish(sequence); // Публикуем
            }
        }

        // Ждем завершения
        Thread.sleep(1000);
        disruptor.shutdown();
    }
}

4. Основные классы и методы

4.1. RingBuffer

Сигнатура: public final class RingBuffer<E> implements Cursored, EventSequencer<E>, EventSink<E>

Что делает: Основной буфер данных фиксированного размера. Хранит события и управляет последовательностями чтения/записи.

RingBuffer<StringEvent> ringBuffer = disruptor.getRingBuffer();
long seq = ringBuffer.next(); // Захватить следующий слот
StringEvent event = ringBuffer.get(seq); // Получить событие
event.setValue("тест");
ringBuffer.publish(seq); // Опубликовать

4.2. EventHandler

Сигнатура: public interface EventHandler<T> { void onEvent(T event, long sequence, boolean endOfBatch) throws Exception; }

Что делает: Интерфейс для обработчиков событий. Вызывается для каждого события.

EventHandler<StringEvent> handler = (event, seq, endOfBatch) -> {
    System.out.println(event.getValue() + " (seq: " + seq + ")");
};
disruptor.handleEventsWith(handler);

4.3. EventFactory

Сигнатура: public interface EventFactory<T> { T newInstance(); }

Что делает: Фабрика для создания пустых событий. Используется для предварительного заполнения RingBuffer.

EventFactory<StringEvent> factory = StringEvent::new;
Disruptor<StringEvent> disruptor = new Disruptor<>(factory, 1024, Executors.defaultThreadFactory());

4.4. Disruptor

Сигнатура: public class Disruptor<T>

Что делает: Главный класс для настройки и управления всей системой Disruptor.

// Создание
Disruptor<StringEvent> disruptor = new Disruptor<>(
    new StringEventFactory(),
    1024,
    DaemonThreadFactory.INSTANCE
);

// Настройка обработчиков
disruptor.handleEventsWith(new StringEventHandler());

// Запуск
disruptor.start();

// Остановка
disruptor.shutdown();

4.5. Sequence

Сигнатура: public class Sequence { public long get(); public void set(long value); public boolean compareAndSet(long expectedValue, long newValue); }

Что делает: Атомарный счетчик последовательности. Используется для отслеживания прогресса чтения/записи.

Sequence sequence = new Sequence(0);
sequence.set(5);
long current = sequence.get();
boolean success = sequence.compareAndSet(5, 10); // true

4.6. SequenceBarrier

Сигнатура: public interface SequenceBarrier { long waitFor(long sequence) throws AlertException, InterruptedException, TimeoutException; }

Что делает: Барьер, который координирует доступ к последовательностям между продюсерами и консумерами.

RingBuffer<StringEvent> ringBuffer = disruptor.getRingBuffer();
SequenceBarrier barrier = ringBuffer.newBarrier();
long nextSequence = barrier.waitFor(10); // Ждать, пока последовательность 10 не станет доступной

4.7. WaitStrategy

Сигнатура: public interface WaitStrategy { long waitFor(long sequence, Sequence cursor, Sequence dependentSequence, SequenceBarrier barrier) throws AlertException, InterruptedException, TimeoutException; }

Что делает: Определяет стратегию ожидания для консумеров (спин-луп, yield, park, блокировка и т.д.).

// Стратегии:
// BusySpinWaitStrategy - максимальная производительность, но жрет CPU
// YieldingWaitStrategy - отдает CPU другим потокам
// SleepingWaitStrategy - спит, если нет событий
// BlockingWaitStrategy - использует блокировки (самая медленная)

Disruptor<StringEvent> disruptor = new Disruptor<>(
    factory, bufferSize, threadFactory,
    ProducerType.SINGLE, // или MULTI
    new YieldingWaitStrategy() // стратегия ожидания
);

4.8. EventHandlerGroup

Сигнатура: public class EventHandlerGroup<T>

Что делает: Позволяет создавать цепочки обработчиков (workflow) с параллельной или последовательной обработкой.

EventHandler<StringEvent> handler1 = (e, s, b) -> System.out.println("1: " + e.getValue());
EventHandler<StringEvent> handler2 = (e, s, b) -> System.out.println("2: " + e.getValue());
EventHandler<StringEvent> handler3 = (e, s, b) -> System.out.println("3: " + e.getValue());

// handler1 и handler2 работают параллельно, затем handler3
disruptor.handleEventsWith(handler1, handler2)
         .then(handler3);

4.9. WorkHandler и WorkerPool

Сигнатура: public interface WorkHandler<T> { void onEvent(T event) throws Exception; }

Что делает: Для сценариев «пул рабочих» (worker pool), где каждое событие обрабатывается только одним из нескольких консумеров.

WorkHandler<StringEvent> worker = event -> {
    System.out.println("Обработано: " + event.getValue());
};

disruptor.handleEventsWithWorkerPool(worker, worker); // 2 рабочих потока

4.10. EventTranslator

Сигнатура: public interface EventTranslator<T> { void translateTo(T event, long sequence); }

Что делает: Удобный способ публикации событий без ручного захвата последовательности.

ringBuffer.publishEvent((event, sequence) -> {
    event.setValue("Автоматическое событие #" + sequence);
});

// Или с дополнительными аргументами
ringBuffer.publishEvent((event, sequence, arg) -> {
    event.setValue(arg);
}, "Привет, мир!");

5. Реальный практический пример

Создадим систему обработки финансовых транзакций. Продюсер генерирует транзакции, два консумера обрабатывают их параллельно: один проверяет баланс, другой записывает в лог. Затем третий консумер агрегирует результаты.

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

// Класс события
class TransactionEvent {
    private long transactionId;
    private double amount;
    private String account;
    private boolean approved;
    private String error;

    // Геттеры и сеттеры
    public long getTransactionId() { return transactionId; }
    public void setTransactionId(long transactionId) { this.transactionId = transactionId; }
    public double getAmount() { return amount; }
    public void setAmount(double amount) { this.amount = amount; }
    public String getAccount() { return account; }
    public void setAccount(String account) { this.account = account; }
    public boolean isApproved() { return approved; }
    public void setApproved(boolean approved) { this.approved = approved; }
    public String getError() { return error; }
    public void setError(String error) { this.error = error; }

    @Override
    public String toString() {
        return String.format("TX#%d: %s %.2f [%s] %s",
            transactionId, account, amount,
            approved ? "APPROVED" : "REJECTED",
            error != null ? error : "");
    }
}

// Фабрика
class TransactionEventFactory implements EventFactory<TransactionEvent> {
    @Override
    public TransactionEvent newInstance() {
        return new TransactionEvent();
    }
}

// Консумер 1: Проверка баланса
class BalanceChecker implements EventHandler<TransactionEvent> {
    private final AtomicInteger processed = new AtomicInteger(0);

    @Override
    public void onEvent(TransactionEvent event, long sequence, boolean endOfBatch) {
        // Симуляция проверки баланса
        if (event.getAmount() > 10000) {
            event.setApproved(false);
            event.setError("Превышен лимит транзакции");
        } else if (event.getAmount() < 0) {
            event.setApproved(false);
            event.setError("Отрицательная сумма");
        } else {
            event.setApproved(true);
        }
        processed.incrementAndGet();
    }

    public int getProcessed() { return processed.get(); }
}

// Консумер 2: Логирование
class TransactionLogger implements EventHandler<TransactionEvent> {
    private final AtomicInteger logged = new AtomicInteger(0);

    @Override
    public void onEvent(TransactionEvent event, long sequence, boolean endOfBatch) {
        System.out.println("[LOG] " + event);
        logged.incrementAndGet();
    }

    public int getLogged() { return logged.get(); }
}

// Консумер 3: Агрегатор (запускается после первых двух)
class Aggregator implements EventHandler<TransactionEvent> {
    private final AtomicInteger approved = new AtomicInteger(0);
    private final AtomicInteger rejected = new AtomicInteger(0);

    @Override
    public void onEvent(TransactionEvent event, long sequence, boolean endOfBatch) {
        if (event.isApproved()) {
            approved.incrementAndGet();
        } else {
            rejected.incrementAndGet();
        }
    }

    public int getApproved