28

I have the classic problem of a thread pushing events to the incoming queue of a second thread. Only this time, I am very interested about performance. What I want to achieve is:

  • I want concurrent access to the queue, the producer pushing, the receiver poping.
  • When the queue is empty, I want the consumer to block to the queue, waiting for the producer.

My first idea was to use a LinkedBlockingQueue, but I soon realized that it is not concurrent and the performance suffered. On the other hand, I now use a ConcurrentLinkedQueue, but still I am paying the cost of wait() / notify() on each publication. Since the consumer, upon finding an empty queue, does not block, I have to synchronize and wait() on a lock. On the other part, the producer has to get that lock and notify() upon every single publication. The overall result is that I am paying the cost of sycnhronized (lock) {lock.notify()} in every single publication, even when not needed.

What I guess is needed here, is a queue that is both blocking and concurrent. I imagine a push() operation to work as in ConcurrentLinkedQueue, with an extra notify() to the object when the pushed element is the first in the list. Such a check I consider to already exist in the ConcurrentLinkedQueue, as pushing requires connecting with the next element. Thus, this would be much faster than synchronizing every time on the external lock.

Is something like this available/reasonable?

2
  • Why do you think that java.util.concurrent.LinkedBlockingQueue is not concurrent? I think it is perfectly concurrent, seeing its javadoc and source code. But have no idea about performance.
    – Rorick
    Commented Jul 31, 2009 at 13:01
  • See also stackoverflow.com/questions/1301691/…
    – Vadzim
    Commented Sep 28, 2012 at 5:49

6 Answers 6

13

I think you can stick to java.util.concurrent.LinkedBlockingQueue regardless of your doubts. It is concurrent. Though, I have no idea about its performance. Probably, other implementation of BlockingQueue will suit you better. There's not too many of them, so make performance tests and measure.

3
  • I say this only by observing my throughput reducing roughly 8 times compared to ConcurrentLinkedQueue. I guessed it was locking the whole thing internally just to offer thread safety. You are right though, it might as well be just worse performance compared to ConcurrentLinkedQueue. Which kind of beats the cause ;-)
    – Yiannis
    Commented Jul 31, 2009 at 13:14
  • 1
    Which ever queue you use, you are going to end up using a lock to support the waiting for a new entry. (unless you busy wait) The lock really isn't that expensive (about 0.5 micro-seconds for Lock) so if its a performance problem you may have a problem with your design, e.g. create less tasks/find a way to add less objects to the queue/batch up your work. Commented Aug 1, 2009 at 7:39
  • 7
    As a note, if you want higher throughput, use drainTo() on your LinkedBlcokingQueue, we measured nearly 500% increased throughput on one of our app, instead of take()'ing one and one element off the queue
    – nos
    Commented Mar 2, 2011 at 18:58
6

Similar to this answer https://stackoverflow.com/a/1212515/1102730 but a bit different.. I ended up using an ExecutorService. You can instantiate one by using Executors.newSingleThreadExecutor(). I needed a concurrent queue for reading/writing BufferedImages to files, as well as atomicity with reads and writes. I only need a single thread because the file IO is orders of magnitude faster than the source, net IO. Also, I was more concerned about atomicity of actions and correctness than performance, but this approach can also be done with multiple threads in the pool to speed things up.

To get an image (Try-Catch-Finally omitted):

Future<BufferedImage> futureImage = executorService.submit(new Callable<BufferedImage>() {
    @Override
        public BufferedImage call() throws Exception {
            ImageInputStream is = new FileImageInputStream(file);
            return  ImageIO.read(is);
        }
    })

image = futureImage.get();

To save an image (Try-Catch-Finally omitted):

Future<Boolean> futureWrite = executorService.submit(new Callable<Boolean>() {
    @Override
    public Boolean call() {
        FileOutputStream os = new FileOutputStream(file); 
        return ImageIO.write(image, getFileFormat(), os);  
    }
});

boolean wasWritten = futureWrite.get();

It's important to note that you should flush and close your streams in a finally block. I don't know about how it performs compared to other solutions, but it is pretty versatile.

5

I would suggest you look at ThreadPoolExecutor newSingleThreadExecutor. It will handle keeping your tasks ordered for you, and if you submit Callables to your executor, you will be able to get the blocking behavior you are looking for as well.

4

You can try LinkedTransferQueue from jsr166: http://gee.cs.oswego.edu/cgi-bin/viewcvs.cgi/jsr166/src/jsr166y/

It fulfills your requirements and have less overhead for offer/poll operations. As I can see from the code, when the queue is not empty, it uses atomic operations for polling elements. And when the queue is empty, it spins for some time and park the thread if unsuccessful. I think it can help in your case.

3

I use the ArrayBlockingQueue whenever I need to pass data from one thread to another. Using the put and take methods (which will block if full/empty).

2

Here is a list of classes implementing BlockingQueue.

I would recommend checking out SynchronousQueue.

Like @Rorick mentioned in his comment, I believe all of those implementations are concurrent. I think your concerns with LinkedBlockingQueue may be out of place.

2
  • 1
    SynchronousQueue is not what I want, as it will block my producer each time that it is trying to publish .
    – Yiannis
    Commented Jul 31, 2009 at 13:05
  • SynchronousQueue looks more like a pipe, than queue. Looks like it cannot contain tasks pending processing in it, only "push" single tasks from producer to consumer.
    – Rorick
    Commented Jul 31, 2009 at 13:07

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.