Take the 2-minute tour ×
Code Review Stack Exchange is a question and answer site for peer programmer code reviews. It's 100% free, no registration required.

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){
    queue = new LinkedList<T>();
    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();
        queue.add(item);
    } 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();
    }
}
share|improve this question

2 Answers 2

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.

share|improve this answer

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();
            }
            queue.add(object);

            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 {
                Thread.sleep(priority);
                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 {
                Thread.sleep(priority);
                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);

        new Thread(new Producer(200, sharedBuffer)).start();
        new Thread(new Producer(300, sharedBuffer)).start();
        new Thread(new Consumer(1000, sharedBuffer)).start();
        new Thread(new Consumer(1000, sharedBuffer)).start();
    }
}
share|improve this answer
    
You should explain the changes you made more thoroughly. Right now this is more of a "code dump" answer. –  syb0rg Aug 10 at 16:05

Your Answer

 
discard

By posting your answer, you agree to the privacy policy and terms of service.

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