-2
\$\begingroup\$

An implementation of BlockingQueue

import java.util.LinkedList;
import java.util.List;

public class MyBlockingQueue<T> {
    List<T> queue = new LinkedList<T>();
    private int limit = 0;

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

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

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

}
\$\endgroup\$
3
\$\begingroup\$

In your constructor you should validate your limit. Limits less than 1 are probably an error.

I mentioned in another answer that you should avoid synchronized methods because that leaves you vulnerable to uncontrolled locking. In this case, you have the private queue instance. You should make that final, and then use that as the monitor-lock for synchronization (the limit should be final too). Hmmm... your queue is not even private. In your current code, it is possible for another class in your package to change the contents of the queue without locking on the controls at all...:

private final List<T> queue = new LinkedList<T>();
private final int limit;

Note how the limit is not initialized here, but in the constructor.

Hmm, why use List<T> when you can use the Deque<T> personality of LinkedList? It should be:

private final Deque<T> queue = new LinkedList<T>();

then use the addLast and removeFirst methods....

You should, if you can, make the whole class final as well, to prevent people from making sub-classes of your queue.

Then, in your methods, synchronize on the queue:

public void enqueue(T item) throws InterruptedException {
    synchronized(queue) { 
        while (queue.size() == limit) {
            queue.wait();
        }
        if (queue.isEmpty()) {
            queue.notify();
        }
        queue.addLast(item);
    }
}

There I use the better isEmpty() method. Additionally, you only need to notify a single waiting thread that there's data (you can do this now because nobody other than you can lock on the queue).

The matching dequeue method would be:

public T dequeue() throws InterruptedException {
    synchronized(queue) {
        while (queue.isEmpty()) {
            queue.wait();
        }
        if (queue.size() == limit) {
            queue.notify();
        }
        return queue.removeFirst();
    }
}
\$\endgroup\$
  • \$\begingroup\$ notify() is not correct, notifyAll() is needed for this convoluted race condition: stackoverflow.com/questions/13774802/… \$\endgroup\$ – Thomas Mathew Jul 21 '15 at 19:25
  • \$\begingroup\$ Hmm... you're right. Two things - 1. I would have found that in testing ;-) 2. I would consider a ReentrantLock with two Conditions, one for full, and one for empty, as per the Condition documentation \$\endgroup\$ – rolfl Jul 21 '15 at 19:44
0
\$\begingroup\$

I recommend using ReentrantLock, it's not easy to use synchronized right.

import java.util.Deque;
import java.util.LinkedList;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

// Reference: http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/locks/Condition.html
public final class MyBlockingQueue<T> {
    private final Lock lock = new ReentrantLock();
    private final Condition notFull  = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();

    private final Deque<T> queue = new LinkedList<T>();
    private final int limit;

    public MyBlockingQueue(int limit) {
        if (limit < 1) throw new IllegalStateException("limit should be greater than 0");
        this.limit = limit;
    }

    public synchronized void enqueue(T item) throws InterruptedException {
        lock.lock();
        try {
            while (queue.size() == limit) {
                notFull.await();
            }
            queue.addLast(item);
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public synchronized T dequeue() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                notEmpty.await();
            }
            final T x = queue.removeFirst();
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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