BlockingQueue的简单实现

Question

一种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);
    }

}

Answer 1

在构造函数中，应该验证限制。小于1的限制可能是一个错误。

我在另一个答案中提到，您应该避免同步方法，因为这会使您容易受到无法控制的锁定。在本例中，您有私有队列实例。您应该将其确定为最终，然后使用它作为同步的监视器锁(限制也应该是最终的)。嗯..。你的队列甚至不是私人的。在当前代码中，包中的另一个类有可能在不锁定控件的情况下更改队列的内容……：

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

注意，这个限制在这里不是初始化的，而是在构造函数中。

嗯，既然你可以使用List<T>的Deque<T>个性，为什么要使用LinkedList呢？它应该是：

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

然后使用addLast和removeFirst方法..。

如果可以的话，你也应该让整个类都是最终的，以防止人们对你的队列做子类。

然后，在您的方法中，在队列上同步：

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

在这里，我使用了更好的isEmpty()方法。此外，您只需要通知单个等待线程有数据(您现在可以这样做，因为除了您可以锁定队列之外，没有其他任何人)。

匹配的去队列方法将是：

public T dequeue() throws InterruptedException {
    synchronized(queue) {
        while (queue.isEmpty()) {
            queue.wait();
        }
        if (queue.size() == limit) {
            queue.notify();
        }
        return queue.removeFirst();
    }
}

Answer 2

我建议使用ReentrantLock，使用synchronized并不容易。

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();
        }
    }
}