# BlockingQueue Implemetation using ReentrantLock

I was writing my own implementation of BlockingQueue for practice. I am trying to avoid using the synchronized keyword for the methods. I would instead like to use ReentrantLock.

What is the best way to write this implementation? I am not a Java ninja and would greatly appreciate if someone could pinpoint the errors in my code here and suggest better ways of implementing it.

public class MyBlockingQueue<T> {

private Queue<T> queue;
private AtomicInteger limit = new AtomicInteger(10);
private Lock put_lock = new ReentrantLock();
private Lock take_lock = new ReentrantLock();
private Condition put_condition = put_lock.newCondition();
private Condition take_condition = take_lock.newCondition();

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

public boolean put(T item) throws InterruptedException{
put_lock.lockInterruptibly();
try {
while(queue.size() == limit.get()) {
put_condition.await();
}
put_condition.signal();
} finally{
put_lock.unlock();
}

return true;
}

public T take() throws InterruptedException{
take_lock.lockInterruptibly();
try {
while (queue.size() == 0) {
take_condition.await();
}
take_condition.signal();
return queue.poll();
} finally {
take_lock.unlock();
}
}

-

These seem mixed up:

    while(queue.size() == limit.get()) {
put_condition.await();
}
put_condition.signal();


and

    while (queue.size() == 0) {
take_condition.await();
}
take_condition.signal();


when putting if queue is full you wait until someone takes and when taking if queue is empty you wait until someone puts.

You should have test cases with 11 puts followed by 1 take. and 1 take followed by 1 put. To see they do not deadlock.

What do you know, the example code for Condition is actually Bounded buffer. And you can see it here: http://docs.oracle.com/javase/6/docs/api/java/util/concurrent/locks/Condition.html

But I like their condition names better;)

Limit is a constant as it is, hence does not need atomic integer or any other synchronization. Just declare it final. best way of making something threadsafe is to make it immutable.

-

I think we can do it the next way (please pay attention, that we can use one Condition)

class BlockingQueue<T> {

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

private final ReentrantLock lock = new ReentrantLock();

private Condition emptyQueue = lock.newCondition();

private Condition fullQueue = lock.newCondition();

private int capacity = 10;

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

public void put(T object) throws InterruptedException {
try {
lock.lock();

while (queue.size() == capacity) {
fullQueue.await();
}
if (queue.size() == 0) {
emptyQueue.signalAll();
}

System.out.println("put  " + queue.size());

} finally {

lock.unlock();
}
}

public T take() throws InterruptedException {
try {
lock.lock();

while (queue.size() == 0) {
emptyQueue.await();
}
if (queue.size() == capacity) {
fullQueue.signalAll();
}
System.out.println("take " + (queue.size() - 1));

return queue.remove(0);

} finally {

lock.unlock();
}
}
}

class Producer implements Runnable {

private final BlockingQueue<Object> queue;

private int priority;

public Producer(int priority, BlockingQueue<Object> queue) {
this.priority = priority;
this.queue = queue;
}

@Override
public void run() {
while (true) {
try {
queue.put("object");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

class Consumer implements Runnable {

private final BlockingQueue<Object> queue;

private int priority;

public Consumer(int priority, BlockingQueue<Object> queue) {
this.priority = priority;
this.queue = queue;
}

@Override
public void run() {
while (true) {
try {
queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}

public class ConsumerProducer {

private static final int SHARED_BUFFER_CAPACITY = 10;

public static void main(String[] args) {

BlockingQueue<Object> sharedBuffer = new BlockingQueue<Object>(SHARED_BUFFER_CAPACITY);