Java Concurrency: Blocking Queues

BIRT Reports Made Simple FREE 90 Day Trial
iServer Express for BIRT eclipse.dzone.com

Free Open Source Project Hosting Disappointed with SourceForge?
Host your project at JavaForge! JavaForge.com

Ads by DZone

A blocking queue is a queue that blocks when you try to dequeue from it and the queue is empty, or if you try to enqueue items to it and the queue is already full. A thread trying to dequeue from an empty queue is blocked until some other thread inserts an item into the queue. A thread trying to enqueue an item in a full queue is blocked until some other thread makes space in the queue, either by dequeuing one or more items or clearing the queue completely.

Java 5 comes with blocking queue implementations in the java.util.concurrent package. Yet it can be useful to know the theory behind their implementation.

Blocking Queue Implementation

The implementation of a blocking queue looks similar to a Bounded Semaphore. Here is a simple implementation of a blocking queue:

public class BlockingQueue {

  private List queue = new LinkedList();
  private int  limit = 10;

  public BlockingQueue(int limit){
    this.limit = limit;
  }


  public synchronized void enqueue(Object item)
  throws InterruptedException  {
    while(this.queue.size() == this.limit) {
      wait();
    }
    if(this.queue.size() == 0) {
      notifyAll();
    }
    this.queue.add(item);
  }


  public synchronized Object dequeue()
  throws InterruptedException{
    while(this.queue.size() == 0){
      wait();
    }
    if(this.queue.size() == this.limit){
      notifyAll();
    }

    return this.queue.remove(0);
  }

}

Notice how notifyAll() is only called from enqueue() and dequeue() if the queue size is equal to the size bounds (0 or limit). If the queue size is not equal to either bound when enqueue() or dequeue() is called, there can be no threads waiting to either enqueue or dequeue items.

This text is part of my tutorial on Java Concurrency which by now contains 19 texts.

Average:

(Note: Opinions expressed in this article and its replies are the opinions of their respective authors and not those of DZone, Inc.)

Comments

pugal replied on Tue, 2008/06/24 - 3:06am

In the dequeue method has to be

public synchronized Object dequeue()
throws InterruptedException{
while(this.queue.size() == 0){ // not while(this.limit == 0)
wait();
}
if(this.queue.size() == this.limit){
notifyAll();
}
return this.queue.remove(0);
}

please check and change

- Pugal

Jakob Jenkov replied on Tue, 2008/06/24 - 4:29am

thanks pugal... a minor glitch :-) ... Fixed !

Steven Vetzal replied on Tue, 2008/06/24 - 7:50am

Interesting, but I'd encourage developers that actually need such a mechanism to use Doug Lee's util-concurrent package, or java.util.concurrent.BlockingQueue in Java 1.5+ instead of writing their own. Concurrency is a very tricky topic to get right, very easy to unearth deadlocks or state problems in what seems to be bug free code.

Steven Vetzal replied on Tue, 2008/06/24 - 7:57am

Because your two different synchronized methods could each be called concurrently you would probably be best to put each method in a synchronzed block on your List object...

public void enqueue(Object item) {

  synchronized(queue) {

...

public Object dequeue() {

Jakob Jenkov replied on Tue, 2008/06/24 - 8:06am

Agreed. The second paragraph also mentions the Java 5 implementation. In fact the Java 5 blocking queue is implemented by Doug Lea too. Many of the Java 5 concurrency utils are implemented by Doug Lea. At least I have seen his name is in some of the sources.

Jakob Jenkov replied on Tue, 2008/06/24 - 8:11am

What difference would synchronizing on the List object give compared to synchronizing on "this"? I can't really think of any. The methods could still not be called concurrently. The synchronized(queue) would still block all other threads calling either method, while a thread is executing inside one of them.

Steven Vetzal replied on Tue, 2008/06/24 - 9:34am

You're absolutely right - I had always thought individually marked synchronized methods could be called concurrently. I dug up this resource from Sun looking to prove or disprove either of our positions- no offense intended)

http://java.sun.com/docs/books/tutorial/essential/concurrency/syncmeth.html

Doug Lea's concurrency work is pervasive within the Java community, he's reasoned out this animal pretty thoroughly.

Thanks Jakob - I appreciate the correction.

Slava Imeshev replied on Tue, 2008/06/24 - 8:26pm in response to: jj83777

jj83777 wrote:

What difference would synchronizing on the List object give compared to synchronizing on "this"? I can't really think of any.

There is a huge difference. In the first case the monitor is the list object and the second is this.

The methods could still not be called concurrently. The synchronized(queue) would still block all other threads calling either method, while a thread is executing inside one of them.

If you use list and then someone uses synchronized qualifier on some other method thinking that he owns the monitor while he is not, this can lead to very hard to track bugs because threads will be accessing the shared data concurrently. It's better to start with a simple solution, which was your original code with synchronized methods :-)

Regards,

Slava Imeshev

Cacheonix - Thread-safe Distributed Hashtable

pugal replied on Wed, 2008/06/25 - 1:35am

Hi Jakob

J2ME doesn't have locks and blocking queues kind of stuff.. I really enjoy your work, it gives me the options to work with locks and other utill.concurrent package classes in Java ME. Continue your good work.. cheers

Thanks and Regards

Taylor Gautier replied on Wed, 2008/06/25 - 2:32am

I realize it's probably out of scope of your article, but you should probably mention that most efficient blocking queue implementations (including Java 5's LinkedBlockingQueue) implement a two-lock mechanism that reduces contention between producers and consumers.

Jakob Jenkov replied on Wed, 2008/06/25 - 3:54am

@Taylor Gautier
Good point. I am sure you could optimize the design for performance. You probably could with many of the constructs I have described in my tutorial. But the purpose of the tutorial is first of all to show the basic principles. To show things that work. Then later perhaps I can add the optimizations (if and when I learn about them). My priorities for software and texts is usually:

1) Make it work
2) Make it right
3) Make it fast

@Slava
I know there is a difference between synchronized(this) and synchronized(queue). What I meant was that the BlockingQueue implementation shown here would not be more concurrent by changing from synchronized(this) to synchronized(queue). It would have to use two locks (one for readers and one for writers) to be more concurrent, as Taylor mentions.

@Steven Vetzal
No offense taken :-) I get corrected all the time when I write these texts. Sometimes minor corrections, sometimes major corrections. I learn a lot from these corrections and I apply them to the texts as soon as I have time, so the text is correct in the future. For instance, David Karr and Slava pointed out errors in my reentrant implementation of ReadWriteLock posted 1-2 weeks ago. I learned from that and fixed the text. But sometimes people write "corrections" that are not "true", so I am always a bit critical/skeptical until I fully understand what people mean. I didn't mean to offend you either with my comment. I just wanted to know if you knew of some difference that I didn't :-) Then I'd learn something new. Again :-)

Jakob Jenkov replied on Wed, 2008/06/25 - 3:58am

BTW if anyone have the code for optimized implementations, how about a follow-up post here on JavaLobby? I wouldn't mind seeing more efficient versions of the simple implementations I show in my texts.