Package com.lmax.disruptor

package com.lmax.disruptor

The Disruptor is a concurrent programming framework for exchanging and coordinating work as a continuous series of events. It can be used as an alternative to wiring processing stages together via queues. The Disruptor design has the characteristics of generating significantly less garbage than queues and separates the concurrency concerns so non-locking algorithms can be employed resulting in greater scalability and performance.

It works on the principle of having a number of stages that are each single threaded with local state and memory. No global memory exists and all communication is achieved by passing messages/state via managed ring buffers.

Almost any graph or pipeline structure can be composed via one or more Disruptor patterns.

UniCast a series of items between 1 publisher and 1 EventProcessor.

                                           track to prevent wrap
                                           +------------------+
                                           |                  |
                                           |                  v
 +----+    +-----+            +----+    +====+    +====+   +-----+
 | P1 |--->| EP1 |            | P1 |--->| RB |<---| SB |   | EP1 |
 +----+    +-----+            +----+    +====+    +====+   +-----+
                                   claim      get    ^        |
                                                     |        |
                                                     +--------+
                                                       waitFor
 

Sequence a series of messages from multiple publishers

                                          track to prevent wrap
                                          +--------------------+
                                          |                    |
                                          |                    v
 +----+                       +----+    +====+    +====+    +-----+
 | P1 |-------+               | P1 |--->| RB |<---| SB |    | EP1 |
 +----+       |               +----+    +====+    +====+    +-----+
              v                           ^   get    ^         |
 +----+    +-----+            +----+      |          |         |
 | P2 |--->| EP1 |            | P2 |------+          +---------+
 +----+    +-----+            +----+      |            waitFor
              ^                           |
 +----+       |               +----+      |
 | P3 |-------+               | P3 |------+
 +----+                       +----+
 

Pipeline a series of messages

                           +----+    +-----+    +-----+    +-----+
                           | P1 |--->| EP1 |--->| EP2 |--->| EP3 |
                           +----+    +-----+    +-----+    +-----+



                           track to prevent wrap
              +----------------------------------------------------------------+
              |                                                                |
              |                                                                v
 +----+    +====+    +=====+    +-----+    +=====+    +-----+    +=====+    +-----+
 | P1 |--->| RB |    | SB1 |<---| EP1 |<---| SB2 |<---| EP2 |<---| SB3 |<---| EP3 |
 +----+    +====+    +=====+    +-----+    +=====+    +-----+    +=====+    +-----+
      claim   ^  get    |   waitFor           |   waitFor           |  waitFor
              |         |                     |                     |
              +---------+---------------------+---------------------+
 

Multicast a series of messages to multiple EventProcessors

           +-----+                                        track to prevent wrap
    +----->| EP1 |                        +--------------------+----------+----------+
    |      +-----+                        |                    |          |          |
    |                                     |                    v          v          v
 +----+    +-----+            +----+    +====+    +====+    +-----+    +-----+    +-----+
 | P1 |--->| EP2 |            | P1 |--->| RB |<---| SB |    | EP1 |    | EP2 |    | EP3 |
 +----+    +-----+            +----+    +====+    +====+    +-----+    +-----+    +-----+
    |                              claim      get    ^         |          |          |
    |      +-----+                                   |         |          |          |
    +----->| EP3 |                                   +---------+----------+----------+
           +-----+                                                 waitFor
 

Replicate a message then fold back the results

           +-----+                               track to prevent wrap
    +----->| EP1 |-----+                   +-------------------------------+
    |      +-----+     |                   |                               |
    |                  v                   |                               v
 +----+             +-----+   +----+    +====+               +=====+    +-----+
 | P1 |             | EP3 |   | P1 |--->| RB |<--------------| SB2 |<---| EP3 |
 +----+             +-----+   +----+    +====+               +=====+    +-----+
    |                  ^           claim   ^  get               |   waitFor
    |      +-----+     |                   |                    |
    +----->| EP2 |-----+                +=====+    +-----+      |
           +-----+                      | SB1 |<---| EP1 |<-----+
                                        +=====+    +-----+      |
                                           ^                    |
                                           |       +-----+      |
                                           +-------| EP2 |<-----+
                                          waitFor  +-----+
 

Code Example

 // Event holder for data to be exchanged
 public final class ValueEvent
 {
     private long value;

     public long getValue()
     {
         return value;
     }

     public void setValue(final long value)
     {
         this.value = value;
     }

     public final static EventFactory<ValueEvent> EVENT_FACTORY = new EventFactory<ValueEvent>()
     {
         public ValueEvent newInstance()
         {
             return new ValueEvent();
         }
     };
 }

 // Callback handler which can be implemented by EventProcessors
 final EventHandler<ValueEvent> eventHandler = new EventHandler<ValueEvent>()
 {
     public void onEvent(final ValueEvent event, final long sequence, final boolean endOfBatch)
         throws Exception
     {
         // process a new event as it becomes available.
     }
 };

 RingBuffer<ValueEvent> ringBuffer =
     new RingBuffer<ValueEvent>(ValueEvent.EVENT_FACTORY,
                                new SingleThreadedClaimStrategy(BUFFER_SIZE),
                                new SleepingWaitStrategy());

 SequenceBarrier<ValueEvent> sequenceBarrier = ringBuffer.newBarrier();
 BatchEventProcessor<ValueEvent> batchProcessor = new BatchEventProcessor<ValueEvent>(sequenceBarrier, eventHandler);
 ringBuffer.setGatingSequences(batchProcessor.getSequence());

 // Each processor runs on a separate thread
 EXECUTOR.submit(batchProcessor);

 // Publishers claim events in sequence
 long sequence = ringBuffer.next();
 ValueEvent event = ringBuffer.get(sequence);

 event.setValue(1234);

 // publish the event so it is available to EventProcessors
 ringBuffer.publish(sequence);
 

Related Packages

Package	Description
com.lmax.disruptor.dsl	A DSL-style API for setting up the disruptor pattern around a ring buffer.
com.lmax.disruptor.util	Utility classes

Class	Description
AbstractSequencer	Base class for the various sequencer types (single/multi).
AggregateEventHandler<T>	An aggregate collection of EventHandlers that get called in sequence for each event.
AlertException	Used to alert EventProcessors waiting at a SequenceBarrier of status changes.
BatchEventProcessor<T>	Convenience class for handling the batching semantics of consuming entries from a RingBuffer and delegating the available events to an EventHandler.
BatchEventProcessor.NoRewindHandler
BatchEventProcessorBuilder
BatchRewindStrategy	Strategy for handling a rewindableException when processing an event.
BlockingWaitStrategy	Blocking strategy that uses a lock and condition variable for EventProcessors waiting on a barrier.
BusySpinWaitStrategy	Busy Spin strategy that uses a busy spin loop for EventProcessors waiting on a barrier.
Cursored	Implementors of this interface must provide a single long value that represents their current cursor value.
DataProvider<T>	Typically used to decouple classes from RingBuffer to allow easier testing
EventFactory<T>	Called by the RingBuffer to pre-populate all the events to fill the RingBuffer.
EventHandler<T>	Callback interface to be implemented for processing events as they become available in the RingBuffer
EventHandlerBase<T>
EventHandlerIdentity
EventPoller<T>	Experimental poll-based interface for the Disruptor.
EventPoller.Handler<T>	A callback used to process events
EventPoller.PollState	Indicates the result of a call to EventPoller.poll(Handler)
EventProcessor	An EventProcessor needs to be an implementation of a runnable that will poll for events from the RingBuffer using the appropriate wait strategy.
EventSequencer<T>	Pulls together the low-level data access and sequencing operations of RingBuffer
EventSink<E>	Write interface for RingBuffer.
EventTranslator<T>	Implementations translate (write) data representations into events claimed from the RingBuffer.
EventTranslatorOneArg<T,A>	Implementations translate another data representations into events claimed from the RingBuffer
EventTranslatorThreeArg<T,A,B,C>	Implementations translate another data representations into events claimed from the RingBuffer
EventTranslatorTwoArg<T,A,B>	Implementations translate another data representations into events claimed from the RingBuffer
EventTranslatorVararg<T>	Implementations translate another data representations into events claimed from the RingBuffer
EventuallyGiveUpBatchRewindStrategy	Strategy for handling a rewindableException that will eventually delegate the exception to the ExceptionHandler after a specified number of attempts have been made.
ExceptionHandler<T>	Callback handler for uncaught exceptions in the event processing cycle of the BatchEventProcessor
ExceptionHandlers	Provides static methods for accessing a default ExceptionHandler object.
ExceptionHandlers.DefaultExceptionHandlerHolder
FatalExceptionHandler	Convenience implementation of an exception handler that uses the standard JDK logging of System.Logger to log the exception as System.Logger.Level.ERROR and re-throw it wrapped in a RuntimeException
FixedSequenceGroup	Hides a group of Sequences behind a single Sequence
IgnoreExceptionHandler	Convenience implementation of an exception handler that uses the standard JDK logging of System.Logger to log the exception as System.Logger.Level.INFO
InsufficientCapacityException	Exception thrown when it is not possible to insert a value into the ring buffer without it wrapping the consuming sequences.
LhsPadding
LifecycleAware	Implement this interface in your EventHandler to be notified when a thread for the BatchEventProcessor starts and shuts down.
LiteBlockingWaitStrategy	Variation of the BlockingWaitStrategy that attempts to elide conditional wake-ups when the lock is uncontended.
LiteTimeoutBlockingWaitStrategy	Variation of the TimeoutBlockingWaitStrategy that attempts to elide conditional wake-ups when the lock is uncontended.
MultiProducerSequencer	Coordinator for claiming sequences for access to a data structure while tracking dependent Sequences.
NanosecondPauseBatchRewindStrategy	Strategy for handling a rewindableException that will pause for a specified amount of nanos.
NoOpEventProcessor	No operation version of a EventProcessor that simply tracks a Sequence.
NoOpEventProcessor.SequencerFollowingSequence	Sequence that follows (by wrapping) another sequence
PhasedBackoffWaitStrategy	Phased wait strategy for waiting EventProcessors on a barrier.
ProcessingSequenceBarrier	SequenceBarrier handed out for gating EventProcessors on a cursor sequence and optional dependent EventProcessor(s), using the given WaitStrategy.
RewindableEventHandler<T>	Callback interface to be implemented for processing events as they become available in the RingBuffer with support for throwing a RewindableException when an even cannot be processed currently but may succeed on retry.
RewindableException	A special exception that can be thrown while using the BatchEventProcessor.
RewindAction	The result returned from the BatchRewindStrategy that decides whether to rewind or throw the exception
RewindHandler
RhsPadding
RingBuffer<E>	Ring based store of reusable entries containing the data representing an event being exchanged between event producer and EventProcessors.
RingBufferFields<E>
RingBufferPad
Sequence	Concurrent sequence class used for tracking the progress of the ring buffer and event processors.
SequenceBarrier	Coordination barrier for tracking the cursor for publishers and sequence of dependent EventProcessors for processing a data structure
Sequenced	Operations related to the sequencing of items in a RingBuffer.
SequenceGroup	A Sequence group that can dynamically have Sequences added and removed while being thread safe.
SequenceGroups	Provides static methods for managing a SequenceGroup object.
Sequencer	Coordinates claiming sequences for access to a data structure while tracking dependent Sequences
SequenceReportingEventHandler<T>	Used by the BatchEventProcessor to set a callback allowing the EventHandler to notify when it has finished consuming an event if this happens after the EventHandler.onEvent(Object, long, boolean) call.
SimpleBatchRewindStrategy	Batch rewind strategy that always rewinds
SingleProducerSequencer	Coordinator for claiming sequences for access to a data structure while tracking dependent Sequences.
SingleProducerSequencer.ProducerThreadAssertion	Only used when assertions are enabled.
SingleProducerSequencerFields
SingleProducerSequencerPad
SleepingWaitStrategy	Sleeping strategy that initially spins, then uses a Thread.yield(), and eventually sleep (LockSupport.parkNanos(n)) for the minimum number of nanos the OS and JVM will allow while the EventProcessors are waiting on a barrier.
TimeoutBlockingWaitStrategy	Blocking strategy that uses a lock and condition variable for EventProcessors waiting on a barrier.
TimeoutException	Wait strategies may throw this Exception to inform callers that a message has not been detected within a specific time window.
Value
WaitStrategy	Strategy employed for making EventProcessors wait on a cursor Sequence.
YieldingWaitStrategy	Yielding strategy that uses a Thread.yield() for EventProcessors waiting on a barrier after an initially spinning.