This has been discussed in stackoverflow
Having a thread-safe backend collection does not necessarily make a
correct program. When only your add method is synchronized, the
take() method may run concurrently to it, so it is possible that
after your if(blockingQueue.remainingCapacity() <= 0) test within
add, a concurrently running take() removes an element, so the
poll() within add may remove an element unnecessarily. There is a
perceivable difference to the situation where add() would complete
before the take(), as the consuming thread would receive a different
item. It other words, the effect would be as if add would sometimes
not remove the oldest item, but the second oldest one.
On the other hand, if you use synchronized for all of your methods
consistently, there is no need to have a thread-safe backend
collection:
import java.util.ArrayDeque;
public class CircularNonBlockingQueue<E> {
private final ArrayDeque<E> blockingQueue;
private final int maxSize;
public CircularNonBlockingQueue(int size) {
if(size<1) throw new IllegalArgumentException("size == "+size);
blockingQueue = new ArrayDeque<>(size);
maxSize = size;
}
public synchronized int size() {
return blockingQueue.size();
}
public synchronized void add(E element) {
if(blockingQueue.size() == maxSize) {
blockingQueue.poll();
}
blockingQueue.add(element);
notify();
}
public synchronized E take() throws InterruptedException {
while(blockingQueue.isEmpty()) wait();
return blockingQueue.remove();
}
}
However, if you can live with weaker guarantees regarding the oldest
element, you can use a BlockingQueue and don’t need any
synchronized:
public class CircularNonBlockingQueue<E> {
private final ArrayBlockingQueue<E> blockingQueue;
public CircularNonBlockingQueue(int size) {
blockingQueue = new ArrayBlockingQueue<>(size);
}
public int size() {
return blockingQueue.size();
}
public void add(E element) {
while(!blockingQueue.offer(element)) {
blockingQueue.poll();
}
}
public E take() throws InterruptedException {
return blockingQueue.take();
}
}
It must be noted that neither of these solutions provides “fairness”.
So if the number of producer and consumer threads is large compared to
the queue’s capacity, there is the risk that producers repeatedly
remove items without reactivating threads blocked in take(). So you
should always ensure to have a sufficiently large capacity.
