Overview

MentaQueue is a super-fast, garbage-less, lock-free, two-thread (producer-consumer) queue based on the Disruptor ideas. Its main characteristics are:

• Memory Barrier: Instead of locking, it uses memory barriers such as the volatile keyword and atomic variables.
• Pooling: It uses a circular array filled with preallocated transfer objects that you can use to send data from one side to the other without having to allocate a new queue entry object for every offer.
• Batching: So that you don't have to hit the memory barrier on every poll, but just when you calculate the max available entries that can be polled. After that you can just poll in a loop without blocking.
• Semi-Volatile Writes: Support for the little known AtomicLong.lazySet() for extra low-latency.

Performance

The tests were executed using an i7 machine overclocked to 4.50Ghz. Full specs here.

Message Throughput:

A producer using an AtomicQueue was able to send 165 million messages in one second to the consumer. That time includes the time it takes for the consumer to receive all the messages. That's 11 times faster than a ConcurrentLinkedQueue running on the same machine. Or 33 times faster than a LinkedBlockingQueue. It is still 38 times faster than an ArrayBlockingQueue.

Producer Latency:

Measuring only the latency in the producer side, in other words, the time it takes for the producer to place one message in the queue, the producer was able to send one message in an average of 23 nanoseconds. That's 2 times faster than a ConcurrentLinkedQueue.

You can read an article about asynchronous logging here.

Message Latency:

Measuring the inter-thread communication latency, in other words, the time it takes to send a single and isolated message from producer to consumer, the producer was able to deliver a long value (8 bytes) to the consumer on an average of 51 nanoseconds. That's 2 times faster than a ConcurrentLinkedQueue. (Note: The producer and the consumer were running in the same core through hyperthreading. Running them in separate cores gives an average latency of 85 nanoseconds)

You can read an article about inter-thread communication here.

Zero Garbage:

It is also worth noticing that the AtomicQueue produces ZERO garbage while the ConcurrentLinkedQueue, the LinkedBlockingQueue and the ArrayBlockingQueue generate garbage even when you insert and remove few elements.

You can check the Benchmarks section in the left column for the full test results.

Sample Code

Below is a basic example of how to use MentaQueue:


package org.mentaqueue.test.sample;

import org.mentaqueue.AtomicQueue;
import org.mentaqueue.BatchingQueue;
import org.mentaqueue.util.Builder;

public class Example {
	
	public static void main(String[] args) {
		
		final BatchingQueue<StringBuilder> queue = new AtomicQueue<StringBuilder>(1024, new Builder<StringBuilder>() {
			@Override
            public StringBuilder newInstance() {
	            return new StringBuilder(1024);
            }
		});
		
		Thread producer = new Thread(new Runnable() {
			
			private final StringBuilder getStringBuilder() {
				StringBuilder sb;
				while((sb = queue.nextToDispatch()) == null) {
					// queue can be full if the size of the queue
					// is small and/or the consumer is too slow
					
					// busy spin (you can also use a wait strategy instead)
				}
				return sb;
			}
			
			@Override
			public void run() {
				
				StringBuilder sb;
				
				while(true) { // the main loop of the thread

					// (...) do whatever you have to do here...
					
					// and whenever you want to send a message to
					// the other thread you can just do:
					sb = getStringBuilder();
					sb.setLength(0);
					sb.append("Hello!");
					queue.flush();
					
					// you can also send in batches to increase throughput:
					sb = getStringBuilder();
					sb.setLength(0);
					sb.append("Hi!");

					sb = getStringBuilder();
					sb.setLength(0);
					sb.append("Hi again!");
					
					queue.flush(); // dispatch the two messages above...
				}
			}
		}, "Thread-Producer");
		
		Thread consumer = new Thread(new Runnable() {

			@Override
			public void run() {
				
				while (true) { // the main loop of the thread
					
					// (...) do whatever you have to do here...
					
					// and whenever you want to check if the producer
					// has sent a message you just do:
					
					long avail;
					while((avail = queue.availableToPoll()) == 0) {
						// queue can be empty!
						// busy spin (you can also use a wait strategy instead)
					}

					for(int i = 0; i < avail; i++) {
						StringBuilder sb = queue.poll();
						// (...) do whatever you want to do with the data
						// just don't call toString() to create garbage...
						// copy byte-by-byte instead...
					}
					queue.donePolling();
				}
			}
		}, "Thread-Consumer");
		
		consumer.start();
		producer.start();
	}
}