The LMAX Disruptor Pattern

The Disruptor is a ring-buffer design for passing events between threads with no locks and no allocation per message. It was built at LMAX, a trading venue, and it extends the SPSC queue from Ring Buffers and Lock-Free Queues with consumer chains and batching. This lesson builds a minimal single-producer Disruptor (project 2) and then maps it to the real library.

Idea	Meaning
Sequence	A padded `long` counter that one thread writes and others read
Cursor	The producer's sequence: the highest published slot
Preallocated events	Every slot holds a mutable event object created once at startup
Sequence barrier	The point a consumer waits at: the minimum of the cursor and the sequences it depends on
Gating sequence	A consumer sequence that stops the producer from overwriting unread slots
Batching	A consumer that falls behind processes every available slot before it publishes its own sequence once

Each example is one complete program. The project examples (2 to 6) repeat the classes from earlier stages in the same file, so every stage compiles and runs on its own; in a real project each class goes in its own file. Example 7 uses the LMAX Disruptor 4.0.0 library.

Design in one table

A queue such as ArrayBlockingQueue stores references to objects that the producer allocates for each message. A Disruptor stores the objects themselves, created once, and producers copy data into them.

Concern	`ArrayBlockingQueue`	Disruptor
Coordination	One `ReentrantLock` and two `Condition`s	Sequences read and written with acquire/release
Allocation per message	One object per message (unless pooled)	None: slots are reused
Consumers	Each message goes to one consumer	Every consumer sees every event (multicast)
Dependencies between consumers	Need a second queue	Barrier on the upstream consumer's sequence
Batching	`drainTo` exists, but each `take` locks	Built in: one sequence write per batch
Waiting	Park on a condition	Pluggable wait strategy: busy spin, yield, park, block

Example 1: preallocated slots and the index mask

The ring size is a power of two, so sequence & (size - 1) gives the slot, the same trick as in Ring Buffers and Lock-Free Queues. Sequences grow forever as long values; only the slot index wraps.

Java

public class Main {
    static final class TradeEvent {
        long id;
        long price;
    }

    public static void main(String[] args) {
        int size = 8;
        int mask = size - 1;
        TradeEvent[] ring = new TradeEvent[size];
        for (int i = 0; i < size; i++) {
            ring[i] = new TradeEvent();
        }
        TradeEvent firstSlot = ring[0];

        for (long seq = 0; seq < 10; seq++) {
            TradeEvent e = ring[(int) (seq & mask)];
            e.id = seq;
            e.price = 100 + seq;
        }

        System.out.println("slot of sequence 9: " + (9 & mask));
        System.out.println("ring[0].id = " + ring[0].id);
        System.out.println("ring[1].id = " + ring[1].id);
        System.out.println("ring[2].id = " + ring[2].id);
        System.out.println("slot 0 is the same object: " + (firstSlot == ring[0]));
    }
}

Output

slot of sequence 9: 1
ring[0].id = 8
ring[1].id = 9
ring[2].id = 2
slot 0 is the same object: true

Sequences 8 and 9 reused slots 0 and 1. No new TradeEvent was created after startup, so the steady state produces no garbage, and the objects stay close together in memory because they were allocated one after another. A real ring must stop the producer before it overwrites a slot a consumer has not read; Example 4 adds that check.

Example 2: a padded Sequence

A sequence is written by one thread and read by others. It uses setRelease and getAcquire from VarHandle, Access Modes, and Fences: a release write publishes every earlier plain write (the event fields), and an acquire read on the other side sees them. Padding keeps two hot sequences out of one 64-byte cache line (false sharing).

The JVM may reorder fields inside one class, so padding fields declared next to value in the same class are not guaranteed to surround it. Fields of a superclass are laid out before fields of a subclass, so the padding goes into a class hierarchy. The LMAX library uses the same layout.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;

class LhsPadding { protected long p1, p2, p3, p4, p5, p6, p7; }
class Value extends LhsPadding { protected volatile long value; }
class RhsPadding extends Value { protected long p9, p10, p11, p12, p13, p14, p15; }

final class Sequence extends RhsPadding {
    private static final VarHandle VALUE;

    static {
        try {
            VALUE = MethodHandles.lookup().findVarHandle(Value.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    public Sequence(long initial) { VALUE.setRelease(this, initial); }

    public long get() { return (long) VALUE.getAcquire(this); }

    public void set(long v) { VALUE.setRelease(this, v); }
}

public class Main {
    public static void main(String[] args) {
        Sequence cursor = new Sequence(-1);
        System.out.println("initial: " + cursor.get());
        cursor.set(41);
        cursor.set(cursor.get() + 1);
        System.out.println("after two sets: " + cursor.get());
    }
}

Output

initial: -1
after two sets: 42

Sequences start at -1, meaning "nothing published yet". The read-then-set on the last line is safe only because one thread owns this sequence. This is the single-writer principle: each sequence has exactly one writer, so no CAS is needed.

Example 3: claim, write, publish

The producer protocol has three steps: claim the next sequence (next), write into the preallocated event (get), and make it visible (publish). Publishing is one release write to the cursor. A consumer that reads the cursor with acquire sees every field written before the publish.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.function.Supplier;

class LhsPadding { protected long p1, p2, p3, p4, p5, p6, p7; }
class Value extends LhsPadding { protected volatile long value; }
class RhsPadding extends Value { protected long p9, p10, p11, p12, p13, p14, p15; }

final class Sequence extends RhsPadding {
    private static final VarHandle VALUE;

    static {
        try {
            VALUE = MethodHandles.lookup().findVarHandle(Value.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    public Sequence(long initial) { VALUE.setRelease(this, initial); }

    public long get() { return (long) VALUE.getAcquire(this); }

    public void set(long v) { VALUE.setRelease(this, v); }
}

final class RingBuffer<E> {
    private final Object[] entries;
    private final int mask;
    private final Sequence cursor = new Sequence(-1);
    private Sequence[] gatingSequences = new Sequence[0];
    private long nextValue = -1;      // producer-only state
    private long cachedGating = -1;   // producer-only state

    public RingBuffer(int size, Supplier<E> factory) {
        if (Integer.bitCount(size) != 1) {
            throw new IllegalArgumentException("size must be a power of two");
        }
        entries = new Object[size];
        for (int i = 0; i < size; i++) {
            entries[i] = factory.get();
        }
        mask = size - 1;
    }

    public void addGatingSequences(Sequence... sequences) { gatingSequences = sequences; }

    public long next() {
        long next = nextValue + 1;
        long wrapPoint = next - entries.length;
        if (wrapPoint > cachedGating) {
            long min;
            while (wrapPoint > (min = minimum(gatingSequences, next - 1))) {
                Thread.onSpinWait();
            }
            cachedGating = min;
        }
        nextValue = next;
        return next;
    }

    public long tryNext() {
        long next = nextValue + 1;
        if (next - entries.length > minimum(gatingSequences, next - 1)) {
            return -1;   // ring is full
        }
        nextValue = next;
        return next;
    }

    @SuppressWarnings("unchecked")
    public E get(long sequence) { return (E) entries[(int) (sequence & mask)]; }

    public void publish(long sequence) { cursor.set(sequence); }

    public Sequence cursor() { return cursor; }

    public SequenceBarrier newBarrier(Sequence... dependents) {
        return new SequenceBarrier(cursor, dependents);
    }

    static long minimum(Sequence[] sequences, long defaultValue) {
        long min = defaultValue;
        for (Sequence s : sequences) {
            min = Math.min(min, s.get());
        }
        return min;
    }
}

// Stub: Example 5 completes the barrier.
final class SequenceBarrier {
    SequenceBarrier(Sequence cursor, Sequence[] dependents) { }
}

public class Main {
    static final class OrderEvent { long orderId; int quantity; }

    public static void main(String[] args) {
        RingBuffer<OrderEvent> ring = new RingBuffer<>(4, OrderEvent::new);

        long seq = ring.next();
        OrderEvent e = ring.get(seq);
        e.orderId = 1001;
        e.quantity = 5;
        System.out.println("claimed " + seq + ", cursor before publish: " + ring.cursor().get());
        ring.publish(seq);
        System.out.println("cursor after publish: " + ring.cursor().get());
        System.out.println("event: " + ring.get(0).orderId + " x" + ring.get(0).quantity);
    }
}

Output

claimed 0, cursor before publish: -1
cursor after publish: 0
event: 1001 x5

Between next and publish the slot is owned by the producer and invisible to consumers, because consumers only read up to the cursor. nextValue and cachedGating are plain fields: only the producer thread touches them.

Example 4: gating stops the producer from wrapping

The producer may claim sequence n only when the slowest gating consumer has passed n - size. Otherwise the producer would overwrite an event that consumer has not read. tryNext reports a full ring instead of spinning, which makes the rule visible in one thread.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.function.Supplier;

class LhsPadding { protected long p1, p2, p3, p4, p5, p6, p7; }
class Value extends LhsPadding { protected volatile long value; }
class RhsPadding extends Value { protected long p9, p10, p11, p12, p13, p14, p15; }

final class Sequence extends RhsPadding {
    private static final VarHandle VALUE;

    static {
        try {
            VALUE = MethodHandles.lookup().findVarHandle(Value.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    public Sequence(long initial) { VALUE.setRelease(this, initial); }

    public long get() { return (long) VALUE.getAcquire(this); }

    public void set(long v) { VALUE.setRelease(this, v); }
}

final class RingBuffer<E> {
    private final Object[] entries;
    private final int mask;
    private final Sequence cursor = new Sequence(-1);
    private Sequence[] gatingSequences = new Sequence[0];
    private long nextValue = -1;      // producer-only state
    private long cachedGating = -1;   // producer-only state

    public RingBuffer(int size, Supplier<E> factory) {
        if (Integer.bitCount(size) != 1) {
            throw new IllegalArgumentException("size must be a power of two");
        }
        entries = new Object[size];
        for (int i = 0; i < size; i++) {
            entries[i] = factory.get();
        }
        mask = size - 1;
    }

    public void addGatingSequences(Sequence... sequences) { gatingSequences = sequences; }

    public long next() {
        long next = nextValue + 1;
        long wrapPoint = next - entries.length;
        if (wrapPoint > cachedGating) {
            long min;
            while (wrapPoint > (min = minimum(gatingSequences, next - 1))) {
                Thread.onSpinWait();
            }
            cachedGating = min;
        }
        nextValue = next;
        return next;
    }

    public long tryNext() {
        long next = nextValue + 1;
        if (next - entries.length > minimum(gatingSequences, next - 1)) {
            return -1;   // ring is full
        }
        nextValue = next;
        return next;
    }

    @SuppressWarnings("unchecked")
    public E get(long sequence) { return (E) entries[(int) (sequence & mask)]; }

    public void publish(long sequence) { cursor.set(sequence); }

    public Sequence cursor() { return cursor; }

    public SequenceBarrier newBarrier(Sequence... dependents) {
        return new SequenceBarrier(cursor, dependents);
    }

    static long minimum(Sequence[] sequences, long defaultValue) {
        long min = defaultValue;
        for (Sequence s : sequences) {
            min = Math.min(min, s.get());
        }
        return min;
    }
}

// Stub: Example 5 completes the barrier.
final class SequenceBarrier {
    SequenceBarrier(Sequence cursor, Sequence[] dependents) { }
}

public class Main {
    static final class OrderEvent { long orderId; }

    public static void main(String[] args) {
        RingBuffer<OrderEvent> ring = new RingBuffer<>(4, OrderEvent::new);
        Sequence consumer = new Sequence(-1);
        ring.addGatingSequences(consumer);

        for (int i = 0; i < 5; i++) {
            long seq = ring.tryNext();
            System.out.println("tryNext -> " + seq);
            if (seq >= 0) {
                ring.publish(seq);
            }
        }
        consumer.set(1);   // the consumer finished sequences 0 and 1
        System.out.println("after consumer reached 1: " + ring.tryNext() + ", " + ring.tryNext() + ", " + ring.tryNext());
    }
}

Output

tryNext -> 0
tryNext -> 1
tryNext -> 2
tryNext -> 3
tryNext -> -1
after consumer reached 1: 4, 5, -1

With size 4 and the consumer at -1, sequence 4 would land on slot 0, which is unread. After the consumer reaches 1, slots 0 and 1 are free, so sequences 4 and 5 succeed and 6 fails. In next the producer caches the gating minimum (cachedGating) and reads the consumer sequences again only when the cached value says the ring might be full. That cache removes most cross-core reads, the same idea as the cached indices in the SPSC queue.

Example 5: barrier, batch processor, and a running pipeline (project 2)

A consumer waits at a sequence barrier. The barrier returns the highest sequence the consumer may read: the minimum of the cursor and every upstream consumer it depends on. The consumer then processes every slot up to that value and writes its own sequence once. This is batching: when the consumer falls behind, it catches up with one release write per batch instead of one per event. The handler gets an endOfBatch flag, which lets it flush IO once per batch.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.function.Supplier;

class LhsPadding { protected long p1, p2, p3, p4, p5, p6, p7; }
class Value extends LhsPadding { protected volatile long value; }
class RhsPadding extends Value { protected long p9, p10, p11, p12, p13, p14, p15; }

final class Sequence extends RhsPadding {
    private static final VarHandle VALUE;

    static {
        try {
            VALUE = MethodHandles.lookup().findVarHandle(Value.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    public Sequence(long initial) { VALUE.setRelease(this, initial); }

    public long get() { return (long) VALUE.getAcquire(this); }

    public void set(long v) { VALUE.setRelease(this, v); }
}

final class RingBuffer<E> {
    private final Object[] entries;
    private final int mask;
    private final Sequence cursor = new Sequence(-1);
    private Sequence[] gatingSequences = new Sequence[0];
    private long nextValue = -1;      // producer-only state
    private long cachedGating = -1;   // producer-only state

    public RingBuffer(int size, Supplier<E> factory) {
        if (Integer.bitCount(size) != 1) {
            throw new IllegalArgumentException("size must be a power of two");
        }
        entries = new Object[size];
        for (int i = 0; i < size; i++) {
            entries[i] = factory.get();
        }
        mask = size - 1;
    }

    public void addGatingSequences(Sequence... sequences) { gatingSequences = sequences; }

    public long next() {
        long next = nextValue + 1;
        long wrapPoint = next - entries.length;
        if (wrapPoint > cachedGating) {
            long min;
            while (wrapPoint > (min = minimum(gatingSequences, next - 1))) {
                Thread.onSpinWait();
            }
            cachedGating = min;
        }
        nextValue = next;
        return next;
    }

    public long tryNext() {
        long next = nextValue + 1;
        if (next - entries.length > minimum(gatingSequences, next - 1)) {
            return -1;   // ring is full
        }
        nextValue = next;
        return next;
    }

    @SuppressWarnings("unchecked")
    public E get(long sequence) { return (E) entries[(int) (sequence & mask)]; }

    public void publish(long sequence) { cursor.set(sequence); }

    public Sequence cursor() { return cursor; }

    public SequenceBarrier newBarrier(Sequence... dependents) {
        return new SequenceBarrier(cursor, dependents);
    }

    static long minimum(Sequence[] sequences, long defaultValue) {
        long min = defaultValue;
        for (Sequence s : sequences) {
            min = Math.min(min, s.get());
        }
        return min;
    }
}

final class SequenceBarrier {
    private final Sequence cursor;
    private final Sequence[] dependents;

    SequenceBarrier(Sequence cursor, Sequence[] dependents) {
        this.cursor = cursor;
        this.dependents = dependents;
    }

    public long waitFor(long sequence) {
        long available;
        while ((available = available()) < sequence) {
            Thread.onSpinWait();   // busy-spin wait strategy
        }
        return available;
    }

    private long available() {
        return RingBuffer.minimum(dependents, cursor.get());
    }
}

interface EventHandler<E> {
    void onEvent(E event, long sequence, boolean endOfBatch);
}

final class BatchEventProcessor<E> implements Runnable {
    private final RingBuffer<E> ringBuffer;
    private final SequenceBarrier barrier;
    private final EventHandler<E> handler;
    private final long lastSequence;          // stop condition for this demo
    private final Sequence sequence = new Sequence(-1);

    public BatchEventProcessor(RingBuffer<E> ringBuffer, SequenceBarrier barrier,
                               EventHandler<E> handler, long lastSequence) {
        this.ringBuffer = ringBuffer;
        this.barrier = barrier;
        this.handler = handler;
        this.lastSequence = lastSequence;
    }

    public Sequence sequence() { return sequence; }

    @Override
    public void run() {
        long next = sequence.get() + 1;
        while (next <= lastSequence) {
            long available = Math.min(barrier.waitFor(next), lastSequence);
            for (long s = next; s <= available; s++) {
                handler.onEvent(ringBuffer.get(s), s, s == available);
            }
            sequence.set(available);   // one release write per batch
            next = available + 1;
        }
    }
}

public class Main {
    static final class ValueEvent { long value; }

    static final class SumHandler implements EventHandler<ValueEvent> {
        long count;
        long sum;
        long batches;

        public void onEvent(ValueEvent e, long sequence, boolean endOfBatch) {
            count++;
            sum += e.value;
            if (endOfBatch) batches++;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        final long events = 1_000_000;
        RingBuffer<ValueEvent> ring = new RingBuffer<>(1024, ValueEvent::new);
        SumHandler handler = new SumHandler();
        BatchEventProcessor<ValueEvent> processor =
                new BatchEventProcessor<>(ring, ring.newBarrier(), handler, events - 1);
        ring.addGatingSequences(processor.sequence());

        Thread consumer = new Thread(processor, "consumer");
        consumer.start();

        for (long v = 1; v <= events; v++) {
            long seq = ring.next();
            ring.get(seq).value = v;
            ring.publish(seq);
        }
        consumer.join();

        System.out.println("count: " + handler.count);
        System.out.println("sum: " + handler.sum);
        System.out.println("batched at least once: " + (handler.batches <= handler.count));
    }
}

Output

count: 1000000
sum: 500000500000
batched at least once: true

The number of batches depends on scheduling, so the program prints only a comparison. The handler fields are read after join, which gives a happens-before edge from the consumer thread to main. The demo stops after a fixed lastSequence; the real library stops processors with an alert flag that waitFor checks.

Example 6: consumer dependencies (a diamond)

LMAX ran a journal stage (write to disk) and a replication stage (send to a backup) in parallel, and a business-logic stage after both. In a Disruptor this needs no extra queues: the business stage's barrier depends on the journal and replication sequences, and the producer gates only on the last stage.

Java

import java.lang.invoke.MethodHandles;
import java.lang.invoke.VarHandle;
import java.util.function.Supplier;

class LhsPadding { protected long p1, p2, p3, p4, p5, p6, p7; }
class Value extends LhsPadding { protected volatile long value; }
class RhsPadding extends Value { protected long p9, p10, p11, p12, p13, p14, p15; }

final class Sequence extends RhsPadding {
    private static final VarHandle VALUE;

    static {
        try {
            VALUE = MethodHandles.lookup().findVarHandle(Value.class, "value", long.class);
        } catch (ReflectiveOperationException e) {
            throw new ExceptionInInitializerError(e);
        }
    }

    public Sequence(long initial) { VALUE.setRelease(this, initial); }

    public long get() { return (long) VALUE.getAcquire(this); }

    public void set(long v) { VALUE.setRelease(this, v); }
}

final class RingBuffer<E> {
    private final Object[] entries;
    private final int mask;
    private final Sequence cursor = new Sequence(-1);
    private Sequence[] gatingSequences = new Sequence[0];
    private long nextValue = -1;      // producer-only state
    private long cachedGating = -1;   // producer-only state

    public RingBuffer(int size, Supplier<E> factory) {
        if (Integer.bitCount(size) != 1) {
            throw new IllegalArgumentException("size must be a power of two");
        }
        entries = new Object[size];
        for (int i = 0; i < size; i++) {
            entries[i] = factory.get();
        }
        mask = size - 1;
    }

    public void addGatingSequences(Sequence... sequences) { gatingSequences = sequences; }

    public long next() {
        long next = nextValue + 1;
        long wrapPoint = next - entries.length;
        if (wrapPoint > cachedGating) {
            long min;
            while (wrapPoint > (min = minimum(gatingSequences, next - 1))) {
                Thread.onSpinWait();
            }
            cachedGating = min;
        }
        nextValue = next;
        return next;
    }

    public long tryNext() {
        long next = nextValue + 1;
        if (next - entries.length > minimum(gatingSequences, next - 1)) {
            return -1;   // ring is full
        }
        nextValue = next;
        return next;
    }

    @SuppressWarnings("unchecked")
    public E get(long sequence) { return (E) entries[(int) (sequence & mask)]; }

    public void publish(long sequence) { cursor.set(sequence); }

    public Sequence cursor() { return cursor; }

    public SequenceBarrier newBarrier(Sequence... dependents) {
        return new SequenceBarrier(cursor, dependents);
    }

    static long minimum(Sequence[] sequences, long defaultValue) {
        long min = defaultValue;
        for (Sequence s : sequences) {
            min = Math.min(min, s.get());
        }
        return min;
    }
}

final class SequenceBarrier {
    private final Sequence cursor;
    private final Sequence[] dependents;

    SequenceBarrier(Sequence cursor, Sequence[] dependents) {
        this.cursor = cursor;
        this.dependents = dependents;
    }

    public long waitFor(long sequence) {
        long available;
        while ((available = available()) < sequence) {
            Thread.onSpinWait();   // busy-spin wait strategy
        }
        return available;
    }

    private long available() {
        return RingBuffer.minimum(dependents, cursor.get());
    }
}

interface EventHandler<E> {
    void onEvent(E event, long sequence, boolean endOfBatch);
}

final class BatchEventProcessor<E> implements Runnable {
    private final RingBuffer<E> ringBuffer;
    private final SequenceBarrier barrier;
    private final EventHandler<E> handler;
    private final long lastSequence;          // stop condition for this demo
    private final Sequence sequence = new Sequence(-1);

    public BatchEventProcessor(RingBuffer<E> ringBuffer, SequenceBarrier barrier,
                               EventHandler<E> handler, long lastSequence) {
        this.ringBuffer = ringBuffer;
        this.barrier = barrier;
        this.handler = handler;
        this.lastSequence = lastSequence;
    }

    public Sequence sequence() { return sequence; }

    @Override
    public void run() {
        long next = sequence.get() + 1;
        while (next <= lastSequence) {
            long available = Math.min(barrier.waitFor(next), lastSequence);
            for (long s = next; s <= available; s++) {
                handler.onEvent(ringBuffer.get(s), s, s == available);
            }
            sequence.set(available);   // one release write per batch
            next = available + 1;
        }
    }
}

public class Main {
    static final class TradeEvent { long value; boolean journaled; }

    static final class Counter implements EventHandler<TradeEvent> {
        long count;
        public void onEvent(TradeEvent e, long seq, boolean end) { count++; }
    }

    public static void main(String[] args) throws InterruptedException {
        final long events = 1_000_000;
        final long last = events - 1;
        RingBuffer<TradeEvent> ring = new RingBuffer<>(1024, TradeEvent::new);

        Counter journalCount = new Counter();
        EventHandler<TradeEvent> journalHandler = (e, seq, end) -> {
            e.journaled = true;              // stage 1 writes into the event
            journalCount.onEvent(e, seq, end);
        };
        Counter replication = new Counter();
        long[] business = new long[2];       // [0] sum, [1] events seen unjournaled
        EventHandler<TradeEvent> businessHandler = (e, seq, end) -> {
            business[0] += e.value;
            if (!e.journaled) business[1]++;
        };

        BatchEventProcessor<TradeEvent> journal =
                new BatchEventProcessor<>(ring, ring.newBarrier(), journalHandler, last);
        BatchEventProcessor<TradeEvent> replicate =
                new BatchEventProcessor<>(ring, ring.newBarrier(), replication, last);
        BatchEventProcessor<TradeEvent> logic = new BatchEventProcessor<>(
                ring, ring.newBarrier(journal.sequence(), replicate.sequence()), businessHandler, last);
        ring.addGatingSequences(logic.sequence());

        Thread[] threads = {
            new Thread(journal), new Thread(replicate), new Thread(logic)
        };
        for (Thread t : threads) t.start();

        for (long v = 1; v <= events; v++) {
            long seq = ring.next();
            TradeEvent e = ring.get(seq);
            e.value = v;
            e.journaled = false;
            ring.publish(seq);
        }
        for (Thread t : threads) t.join();

        System.out.println("journal count: " + journalCount.count);
        System.out.println("replication count: " + replication.count);
        System.out.println("business sum: " + business[0]);
        System.out.println("events not journaled when business saw them: " + business[1]);
    }
}

Output

journal count: 1000000
replication count: 1000000
business sum: 500000500000
events not journaled when business saw them: 0

The last line is 0 because the journal's release write of its sequence happens after it sets journaled, and the business barrier reads that sequence with acquire before it reads the event. Gating on logic alone is enough: logic never passes the two stages it depends on, so its sequence is the minimum of the chain. Every stage reads every event, which a queue cannot do without copying.

Example 7: the same pipeline with the LMAX library

The library (com.lmax:disruptor, version 4.0.0 in this course) packages these pieces behind a DSL. handleEventsWith creates processors fed by the ring, then adds a stage behind them, and publishEvent runs the claim, write, and publish steps with a translator lambda.

Java

import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;

public class Main {
    static final class TradeEvent { long value; }

    static final class SumHandler implements EventHandler<TradeEvent> {
        long sum;
        @Override
        public void onEvent(TradeEvent event, long sequence, boolean endOfBatch) {
            sum += event.value;
        }
    }

    public static void main(String[] args) {
        Disruptor<TradeEvent> disruptor = new Disruptor<>(
                TradeEvent::new, 1024, DaemonThreadFactory.INSTANCE,
                ProducerType.SINGLE, new BusySpinWaitStrategy());

        EventHandler<TradeEvent> journal = (e, seq, end) -> { };
        EventHandler<TradeEvent> replicate = (e, seq, end) -> { };
        SumHandler business = new SumHandler();
        disruptor.handleEventsWith(journal, replicate).then(business);

        RingBuffer<TradeEvent> ring = disruptor.start();
        for (long v = 1; v <= 1_000_000; v++) {
            ring.publishEvent((event, sequence, value) -> event.value = value, v);
        }
        disruptor.shutdown();   // waits until all published events are processed

        System.out.println("business sum: " + business.sum);
    }
}

Output

business sum: 500000500000

ProducerType.SINGLE selects a sequencer without CAS, valid only when one thread publishes. ProducerType.MULTI claims sequences with an atomic add on the cursor and tracks published slots in a separate availability array, because producers can finish writing out of claim order. The translator lambda takes the value as an argument, so it captures nothing and the call allocates nothing per event.

Wait strategies trade latency for CPU:

Strategy	Waiting consumer does	Use when
`BusySpinWaitStrategy`	Spins on the sequence	A core is dedicated to each consumer
`YieldingWaitStrategy`	Spins, then `Thread.yield`	Low latency matters, and spinning threads may share cores with other work
`SleepingWaitStrategy`	Spins, yields, then parks briefly	Latency is less important than CPU use
`BlockingWaitStrategy`	Waits on a lock and condition	CPU must stay free; latency in microseconds is acceptable

Agrona and JCTools

Two other libraries hold lock-free queues used in low-latency Java.

Library	Class names	Shape
JCTools	`SpscArrayQueue`, `MpscArrayQueue`, `SpmcArrayQueue`, `MpmcArrayQueue`, plus linked and chunked variants	`java.util.Queue` implementations; each element goes to one consumer
Agrona	`OneToOneConcurrentArrayQueue`, `ManyToOneConcurrentArrayQueue`, `ManyToManyConcurrentArrayQueue`	Same idea under Agrona's naming
Agrona	`OneToOneRingBuffer`, `ManyToOneRingBuffer`	Binary messages written into a byte buffer, used by Aeron
LMAX Disruptor	`RingBuffer`, `Sequence`, `SequenceBarrier`	Preallocated events, multicast, consumer graphs

The choice depends on the shape of the data flow. A queue fits work distribution, where each item goes to one worker. A Disruptor fits a fixed pipeline where every stage must see every event in order. A byte ring buffer fits messages that are serialized anyway, such as data that leaves the process.

Interview points

The Disruptor replaces locks with sequences: each sequence has a single writer, and readers use acquire reads of release writes.
Events are preallocated; producers copy into a slot, so the steady state allocates nothing.
A barrier returns the minimum of the cursor and upstream consumer sequences; that is how stage dependencies work without extra queues.
Batching: a consumer that falls behind processes all available events and writes its sequence once.
The producer gates on the slowest end consumer and caches that value to avoid reading other cores' cache lines on every claim.
Single-producer mode needs no CAS; multi-producer mode claims with an atomic add and tracks availability per slot.
Padding through superclasses keeps sequences on separate cache lines because the JVM may reorder fields within a class.
Compared with ArrayBlockingQueue: no lock, no per-message allocation, multicast to all consumers, and selectable wait strategies.

Summary

Ring size is a power of two and sequences are ever-growing long values; sequence & mask picks the slot.
Protocol: next, write the event, publish (a release write to the cursor).
Consumers wait at a SequenceBarrier, process a batch, then publish their own sequence.
The producer may claim n only when the slowest gating sequence is at least n - size.
The LMAX library exposes this as Disruptor, handleEventsWith(...).then(...), publishEvent, and wait strategies.
JCTools and Agrona provide array queues named by producer and consumer counts; Agrona also has byte ring buffers.