Java SynchronousQueue

This Java tutorial is to learn about the concurrent collection SynchronousQueue. It is an implementation of BlockingQueue. Among all Java concurrent collections, SynchronousQueue is different. Capacity of a synchrounous queue is always zero. It is because in SynchronousQueue an insert will wait for a remove operation by another thread and vice versa.

• put() call to a SynchronousQueue will not return until there is a corresponding take() call.
• peek is not possible with a SynchronousQueue
• As there is no element iteration is also not possible.
• Insert is not possible if there is a thread trying to remove it.
• SynchronousQueue should be imagined like a baton in a relay race.
• If there are more than one thread waiting for a  removal so that they can do insert then with fairness set to true, threads are granted access in FIFO order.
• SynchronousQueue is the default BlockingQueue used for the Executors.newCachedThreadPool() methods.

SynchronousQueue Example

Let us have a look at a producer-consumer based example to understand the SynchronousQueue.

SynchronousQueueProducer.java

package com.javapapers.java.collections;

import java.util.Random;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;

public class SynchronousQueueProducer implements Runnable {

	protected BlockingQueue<String> blockingQueue;
	final Random random = new Random();

	public SynchronousQueueProducer(BlockingQueue<String> queue) {
		this.blockingQueue = queue;
	}

	@Override
	public void run() {
		while (true) {
			try {
				String data = UUID.randomUUID().toString();
				System.out.println("Put: " + data);
				blockingQueue.put(data);
				Thread.sleep(1000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

SynchronousQueueConsumer.java

package com.javapapers.java.collections;

import java.util.concurrent.BlockingQueue;

public class SynchronousQueueConsumer implements Runnable {

	protected BlockingQueue<String> blockingQueue;

	public SynchronousQueueConsumer(BlockingQueue<String> queue) {
		this.blockingQueue = queue;
	}

	@Override
	public void run() {
		while (true) {
			try {
				String data = blockingQueue.take();
				System.out.println(Thread.currentThread().getName()
						+ " take(): " + data);
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}

}

SynchronousQueueExample.java

package com.javapapers.java.collections;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

public class SynchronousQueueExample {
	public static void main(String[] args) {
		final BlockingQueue<String> synchronousQueue = new SynchronousQueue<String>();

		SynchronousQueueProducer queueProducer = new SynchronousQueueProducer(
				synchronousQueue);
		new Thread(queueProducer).start();

		SynchronousQueueConsumer queueConsumer1 = new SynchronousQueueConsumer(
				synchronousQueue);
		new Thread(queueConsumer1).start();

		SynchronousQueueConsumer queueConsumer2 = new SynchronousQueueConsumer(
				synchronousQueue);
		new Thread(queueConsumer2).start();

	}
}

SynchronousQueueExample Example Output

Put: 4c34138b-51dd-4c2c-9633-87a147253978
Thread-2 take(): 4c34138b-51dd-4c2c-9633-87a147253978
Put: 39682e81-9ba8-4b29-a37e-3dc4d5cdb1f7
Thread-1 take(): 39682e81-9ba8-4b29-a37e-3dc4d5cdb1f7
Put: e21dbfd2-d007-43d2-9733-cb10ce683d81
Thread-2 take(): e21dbfd2-d007-43d2-9733-cb10ce683d81
Put: 556c5584-7154-4117-a0f4-f52ec8dcfad8
Thread-1 take(): 556c5584-7154-4117-a0f4-f52ec8dcfad8
Put: 1da7f16c-c030-44ed-be02-8983b7387497
Thread-2 take(): 1da7f16c-c030-44ed-be02-8983b7387497