Blocking queues

It happens that in our system's architecture we have a situation of concurrent access to some collection. Whether it's a server that handles requests via sockets or a JMS listener, it usually stores incoming objects into a collection and other worker processes them. It can be illustrated as a well known producer-consumer problem.

An example implementation could look like that:

	public class SynchronizedProducer {
	private Queue<String> queue;
	public SynchronizedProducer(Queue queue) {
	this.queue = queue;
	}
	public void produce() throws InterruptedException {
	while (true) {
	putOnQueue(UUID.randomUUID().toString());
	Thread.sleep(1);
	}
	}
	private void putOnQueue(String message) {
	synchronized (queue) {
	queue.add(message);
	}
	}
	}

view raw SynchronizedProducer.java hosted with ❤ by GitHub

We have a class that serves as our producer. In an endless loop we're sending random strings to a queue. Access to it is synchronized by acquiring its monitor. We're sleeping 1 ms so that the processor is not over-killed.

Let's take a look at a consumer:

	public class SynchronizedConsumer {
	private Queue queue;
	public SynchronizedConsumer(Queue queue) {
	this.queue = queue;
	}
	public void consume() throws InterruptedException {
	while (true) {
	takeFromQueue();
	Thread.sleep(1);
	}
	}
	private void takeFromQueue() {
	synchronized (queue) {
	if (!queue.isEmpty()) {
	System.out.println(queue.remove());
	}
	}
	}
	}

view raw SynchronizedConsumer.java hosted with ❤ by GitHub

It synchronizes queue in a similar way. What is more, before taking from the queue, it checks if it's not empty. It also sleeps 1 ms for the same reason as producer.

Class that glues them together:

	public class SynchronizedProducerConsumerApp {
	public static void main(String[] args) throws InterruptedException {
	Queue<String> queue = new LinkedList<String>();
	final SynchronizedConsumer synchronizedConsumer = new SynchronizedConsumer(queue);
	final SynchronizedProducer synchronizedProducer = new SynchronizedProducer(queue);
	Executors.newSingleThreadExecutor().submit(new Runnable() {
	@Override
	public void run() {
	try {
	synchronizedProducer.produce();
	} catch (InterruptedException e) {
	}
	}
	});
	Executors.newSingleThreadExecutor().submit(new Runnable() {
	@Override
	public void run() {
	try {
	synchronizedConsumer.consume();
	} catch (InterruptedException e) {
	}
	}
	});
	}
	}

view raw App.java hosted with ❤ by GitHub

We're using LinkedList as an implementation of our Queue. Producer and consumer are started in a separate threads.

There are at least 3 major drawbacks with this implementation.
First of all we're acquiring monitor of the queue, so we're completely blocking access to it not allowing for a concurrent access.
Secondly we need to check if the queue is not empty before taking from it.
Last but not least, consumer sleeps some specified time (sleeping when you're supposed to be working is not the best thing to do) and our performance decreases.

The remedy for above problems is BlockingQueue.

The new producer is similar to previous one, except that it operates on BlockingQueue and doesn't use any synchronization on it, because it's thread safe.

	public class BlockingQueueProducer {
	private BlockingQueue<String> queue;
	public BlockingQueueProducer(BlockingQueue queue) {
	this.queue = queue;
	}
	public void produce() throws InterruptedException {
	while (true) {
	putOnQueue(UUID.randomUUID().toString());
	Thread.sleep(1);
	}
	}
	private void putOnQueue(String message) {
	queue.add(message);
	}
	}

view raw BlockingQueueProducer.java hosted with ❤ by GitHub

There are some changes in consumer:

	public class BlockingQueueConsumer {
	private BlockingQueue queue;
	public BlockingQueueConsumer(BlockingQueue queue) {
	this.queue = queue;
	}
	public void consume() throws InterruptedException {
	while (true) {
	takeFromQueue();
	}
	}
	private void takeFromQueue() throws InterruptedException {
	System.out.println(queue.take());
	}
	}

view raw BlockingQueueConsumer.java hosted with ❤ by GitHub

Apart from not using synchronization, we don't need to check if the queue is not empty, as method take() is doing it for us. We also don't have to sleep anymore, as it is also handled by this method: as soon as an element is available it will be taken from the queue.

Class to run it:

	public class BlockingQueueConsumerProducerApp {
	public static void main(String[] args) {
	BlockingQueue<String> queue = new LinkedBlockingQueue<String>();
	final BlockingQueueConsumer consumer = new BlockingQueueConsumer(queue);
	final BlockingQueueProducer producer = new BlockingQueueProducer(queue);
	Executors.newSingleThreadExecutor().submit(new Runnable() {
	@Override
	public void run() {
	try {
	producer.produce();
	} catch (InterruptedException e) {
	}
	}
	});
	Executors.newSingleThreadExecutor().submit(new Runnable() {
	@Override
	public void run() {
	try {
	consumer.consume();
	} catch (InterruptedException e) {
	}
	}
	});
	}
	}

view raw BlockingQueueConsumerProducerApp.java hosted with ❤ by GitHub

It hasn't changed much, we're using LinkedBlockingQueue as an implementation.
There are other implementations of BlockingQueue that may be helpful for our needs (e.g. that introduces priority to elements or some delay that is needed before taking).