Java 9 Flow API vs. LMAX Disruptor

Introduction

Java 9 introduces a new class, Flow, that allows developers to take advantage of Reactive programming. Previously, there was a data structure that could do the same thing: LMAX Disruptor.

In this post, I will give a simple example of 2 frameworks to compare syntax and throughput. There are a lot of aspects to be compared, like functionality, memory usage, and initial time. With more complex code, the scenario is limited to 1 publisher and 1 subscriber.

Example Code

Java 9 Flow API

The simplest way to work with Flow API is to define a Subscriber:

    public class SimpleSubscriber < T > implements Subscriber < T > {
 private Subscription subscription;
 public boolean isDone = false;;
 @Override
 public void onSubscribe(Subscription subscription) {
  this.subscription = subscription;
  subscription.request(1);
 }

 @Override
 public void onComplete() {
  isDone = true;
 }

 @Override
 public void onError(Throwable arg0) {
  arg0.printStackTrace();
 }

 @Override
 public void onNext(T arg0) {
  subscription.request(1);
  // do nothing
 }
}

Then init a Publisher and link them together:

            SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();
            EndSubscriber<Integer> subscriber = new EndSubscriber<>();
            publisher.subscribe(subscriber);

Test function:

    @Benchmark
    @BenchmarkMode(Mode.AverageTime)
    @OutputTimeUnit(TimeUnit.MILLISECONDS)
    public void java9(BenchMarkState state) {
        for(int i = 0; i <state.size;i++) {
            state.publisher.submit(state.testData.get(i));
        }
    }

LMAX Disruptor

With LMAX Disruptor, developers have to write more code. Developers have to define their own Event bean, EventFactory<T>, EventHandler<T>, but EventProducer is optional.

Event bean encapsulates all information of an event, so that it depends on dedicated business:

    public class IntegerEvent
    {
        private Integer value;

        public void set(Integer value)
        {
            this.value = value;
        }
        public String toString() {
            return value.toString();
        }
        public void clear() {
            value = null;
        }
    }

Disruptor uses a ring buffer and pre-inited with size of 2^n. The purpose is to reduce overheating of creating new objects and aligning all events in some neighboring fragments of memory, which helps Disruptor travel faster and eliminates memory false sharing. EventFactory will be used to create all Event objects when initializing ring buffer.

    public class IntegerEventFactory implements EventFactory<IntegerEvent>{

        @Override
        public IntegerEvent newInstance() {
            return new IntegerEvent();
        }
    }

EventHandler will get the published Event from the ring buffer, extracting and processing data. In some cases, the data object can live longer than intended. It's better if you have clearing handler.

    public class IntegerEventHandler implements EventHandler<IntegerEvent>
    {
        public void onEvent(IntegerEvent event, long sequence, boolean endOfBatch)
        {
            // do nothing
        }
    }

    public class ClearingEventHandler implements EventHandler<IntegerEvent>
    {
        public void onEvent(IntegerEvent event, long sequence, boolean endOfBatch)
        {
            // Failing to call clear here will result in the 
            // object associated with the event to live until
            // it is overwritten once the ring buffer has wrapped
            // around to the beginning.
            event.clear(); 
        }
    }

EventProducer is optional; developers can get the ring buffer directly from Disruptor and publish new events.

    public static class IntegerEventProducer
    {
        private final RingBuffer<IntegerEvent> ringBuffer;

        public IntegerEventProducer(RingBuffer<IntegerEvent> ringBuffer)
        {
            this.ringBuffer = ringBuffer;
        }

        public void onData(Integer data)
        {
            long sequence = ringBuffer.next();  // Grab the next sequence
            try
            {
                IntegerEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
                // for the sequence
                event.set(data);  // Fill with data
            }
            finally
            {
                ringBuffer.publish(sequence);
            }
        }
    }

Finally, we can link them all together:

            ExecutorService executor = Executors.newCachedThreadPool();
            int bufferSize = 1024;
            Disruptor<IntegerEvent> disruptor = new Disruptor(new IntegerEventFactory(), 
                  bufferSize, executor,  ProducerType.SINGLE, new YieldingWaitStrategy());
            IntegerEventHandler handler = new IntegerEventHandler();
            disruptor.handleEventsWith(handler).then(new ClearingEventHandler());
            disruptor.start();
            RingBuffer<IntegerEvent> ringBuffer = disruptor.getRingBuffer();
            IntegerEventProducer producer = new IntegerEventProducer(ringBuffer);

Test the function:

    @Benchmark
    @BenchmarkMode(Mode.AverageTime)
    @OutputTimeUnit(TimeUnit.MILLISECONDS)
    public void lmaxDisruptor(BenchMarkState state) {        
        for(int i = 0; i <state.size;i++) {
            state.producer.onData(state.testData.get(i));
        }
    }

Syntax Comparison

With the simplest case, Flow API is easier to integrate. Disruptor requires much more code but is more flexible and allows developers to modify more options to adapt to complex business.

Performance Comparison

I used JMH for benchmarking with Intel i7-7700K. The size of data ranges from 5000 items to 50 million items. Each test function will read all items and publish to the stream (or ring buffer), and running time are measured by averages. Lower running time is better in this situation, and we can see LMAX Disruptor runs much faster than Flow API.

Conclusion

LMAX Disruptor has a long history of improvement and optimization so that it performs very well. There is the minor disadvantage that it requires more code, but it's not an everyday task. If you want to study more about LMAX Disruptor, please check the GitHub repo and Martin Fowler's article.

Java 9 Flow API provides nearly the same functionality with fewer lines of code. However, unlike LMAX Disruptor, Flow API doesn't support multiple Publishers—multiple Subcribers scenarios which makes it a big gap between the 2 frameworks.

Start implementing business with Flow API, then migrate to LMAX Disruptor if the business is stable enough, and with good strategy, developers can take advantage of both frameworks.