Java 9 Flow API vs LMAX Disruptor

• Download source code - 60.8 KB

Introduction

Java 9 introduces a new class Flow that allows developers to take advantage of Reactive programming. But far in the past, 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, initial time ... with more complex code, scenario also limited to 1 publisher - 1 subscriber.

Example Code

Java 9 Flow API

The simplest way to work with Flow API is 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 using a ring buffer and pre-inited with size is 2^n. The purpose is to reduce overheating of creating new objects and aligning all events in some neighbouring fragments of memory, help Disruptor travel faster and eliminate 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 published Event from ring buffer, extract and processing data. In some cases, data object can live longer than intended. It's better if 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 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 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 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 more easy to integrate. Disruptor requires much more code but disruptor is more flexible and allows developers to modify more options to adapt with complex business.

Performance Comparison

I use JMH for benchmark with Intel i7-7700K. Size of data from 5000 items to 50 million items. Each test function will read all items and publish to stream (or ring buffer), running time are measured by average. It means lower running time is better. We can see Lmax Disruptor run much faster than Flow API.

Conclusion

Lmax Disruptor has a long history of improve and optimization, so that it performs very well. There is minor disadvantage that it requires more code, but it's not an every day task. If you want to study more about Lmax Disruptor, please check Github repo and Martin Flower's article.

Java 9 Flow API provides nearly the same functionality with few lines of code. Start implementing business with Flow API, then migrate to Lmax Disruptor if business stable enough is good strategy, developers can take advantage of both framework.

The last point is Flow API doesn't support multiple Publishers - multiple Subcribers scenario (Please correct me if I'm wrong), it's a big gap between 2 frameworks.