Hello,
First, keep calm and keep reading :), there is no c/c++ code there, only pure and clear Java code 😀
By disrupting java threads, i mean the way we use them to communicate with each others.
Basically, when you want to enable communication between threads, the first idea is to use a queue : Thread-producer create a workload, put it in a queue to be consumed by Thread-consumer.
Easy to understand, but not efficient if we want to :
- Have multiple consumers : like a multicast queue
- To build a topologie (dependecy graph) can be a nightmare of complexity
- Lock-free : Queues in java are ‘Blocking’ because the use locks, so we can have performance issues there.
So, in this article i will show you how to build a complex inter-thread messaging architecture in an EASY way 🙂
Let me introduce Disruptor framework from LMAX (a trading platform)
Disruptor is a very low-latency and high-throughput inter-thread messaging framework based on standard java framework (no ext lib ^_^) and on the RingBuffer pattern (Circular buffer from wikipedia or more human explanation from Trisha Gee here).
So the Disruptor is more than an implementation of Queue in java, it provides :
- Multicast events to consumers, with consumer dependency graph.
- Pre-allocate memory for events.
- Optionally lock-free.
Here we will use it to implement 3 real work use cases :
Prerequisites
To run the article’s code, you need to have:
– JDK 6 or greater – examples here are in Java 8.
– Disruptor dependency from maven
|
1 2 3 4 5 |
<dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.3.2</version> </dependency> |
or from LMAX repository here
1) Ordered building strategy
We want to have a Strategy that build a car, then set color on it, then set power, then set wheels and finally print it, all that stuff in order.
Easy with the Disruptor DSL : check the code below:
|
1 2 3 4 5 |
disruptor.handleEventsWith(new SetColorCarEventHandler()) .then(new SetPowerCarEventHandler()) .then(new SetWheelCarEventHandler()) .then((CarEvent event, long sequence, boolean endOfBatch) System.out.println(String.format("id [%d] with %s", sequence, event.get()))); |
See method DisruptorTests.buildCarInOrder() for more information.
Here a fully working example with all init:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
public static void main(String[] args) { // 1 - Executor that will be used to construct new threads for consumers ExecutorService executorService = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() / 2); // 2 - Specify the size of the ring buffer, must be power of 2. int bufferSize = 2048; // 3 - initialize the Disruptor object Disruptor<CarEvent> disruptor = new Disruptor(CarEvent::new, bufferSize, executorService);//java 8 flavor disruptor.handleEventsWith(new SetColorCarEventHandler()) .then(new SetPowerCarEventHandler()) .then(new SetWheelCarEventHandler()) .then((CarEvent event, long sequence, boolean endOfBatch) -> System.out.println(String.format("id [%d] with %s", sequence, event.get()))); disruptor.start(); // Get the ring buffer from the Disruptor to be used for publishing. final RingBuffer<CarEvent> ringBuffer = disruptor.getRingBuffer(); for (long l = 0; l < 1000; l++) { // 1 - get next car sequence (sequence is internal ringbuffer counter) long seq = ringBuffer.next(); //2 - get CarEvent object based on it sequence. CarEvent carEvent = ringBuffer.get(seq); //3 - set the payload (Car) in the CarEvent. carEvent.set(new Car()); //4 - publish the event using it sequence. ringBuffer.publish(seq); } } |
See Class ExampleMain for more information.
2) Parallel building strategy
In this strategy we want our car to be Colored, Powered, Wheeled in parallel, and then Print it.
With this strategy, the Same object “Event-A” will be consumed by 3 threads at the same time.
Let see how with the code looks:
|
1 2 3 4 5 |
disruptor.handleEventsWith( new SetColorCarEventHandler(), new SetPowerCarEventHandler(), new SetWheelCarEventHandler()) .then(new PrintCarEventHandler()); |
See method DisruptorTests.buildCarInParallel() for complete example.
3) Dispatching car strategy
After building our car, we want to deliver it by using one delivery type at time (we can’t deliver the same car using multiple ways 😉 ).
For this strategy, we use a WorkPool and define our WorkHandler(business logic executed by each thread) and then build a ringbuffer :
|
1 2 3 4 5 6 7 8 |
final WorkerPool carDeliveryWorkerPool = new WorkerPool(CarEvent::new, new IgnoreExceptionHandler(), new DeliveryCarWorkHandler("Truck"), new DeliveryCarWorkHandler("Ship"), new DeliveryCarWorkHandler("Train")); final RingBuffer ringBuffer = carDeliveryWorkerPool.start(executorService); |
See method DisruptorTests.dispatchEachCarByOnlyOneWarAtOnce() for complete example.
That it, now you can configure a powerful producer/consumer pattern in a easy way using Disruptor api.
Get the code
The code is available on GitHub.
Hope you enjoyed this post, and don’t forget to share!
About the author
Hichame EL KHALFI is a consultant and a senior software architect on financial Front-to-Back software solution.
You can follow him on twitter @Helkhalfi
Links
Disruptor api home page : http://lmax-exchange.github.io/disruptor/
