java - 为什么自定义阻塞队列在 Java 中不是线程安全的

Question

我只是想用ReentrantLock实现一个阻塞队列，我定义了两个条件full和empty，源码如下:

@Slf4j
@NotThreadSafe
public class CustomBlockQueue<T> {

    private ReentrantLock lock = new ReentrantLock();

    private Condition full = lock.newCondition();
    private Condition empty = lock.newCondition();

    private Integer maxLength = 1 << 4;

    private Integer putIndex = 0, takeIndex = 0;
    private Integer count = 0;

    private Object[] value;

    public BlockQueue(){
        value = new Object[maxLength];
    }

    public BlockQueue(Integer maxLength){
        this.maxLength = maxLength;
        value = new Object[maxLength];
    }

    public void put(T val) throws InterruptedException {
        lock.lock();
        try {
            if (count.equals(maxLength)){
                log.info("The queue is full!");
                full.await();
            }
            putIndex = putIndex % maxLength;
            value[putIndex++] = val;
            count++;
            empty.signal();
        }finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lock();
        Object val;
        try {
            if (count == 0){
                empty.await();
            }
            takeIndex = takeIndex % maxLength;
            val = value[takeIndex++];
            count--;
            full.signal();
        }finally {
           lock.unlock();
        }
        return (T) val;
    }
}

在两个消费者线程和一个提供者线程中测试时，count 在某些意外时间小于零。
为什么阻塞队列不是线程安全的，谁能帮帮我，给我一些指导？非常感谢你!

更新(2018/10/17)

如果我只使用一个Condition，它能正常运行吗？源码如下:

@Slf4j
@NotThreadSafe
public class CustomBlockQueue<T> {

    private ReentrantLock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    ...

    public void put(T val) throws InterruptedException {
        lock.lock();
        try {
            while (count.equals(maxLength)){
                log.info("The queue is full!");
                condition.await();
            }
            putIndex = putIndex % maxLength;
            value[putIndex++] = val;
            count++;
            condition.signal();
        }finally {
            lock.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T take() throws InterruptedException {
        lock.lock();
        Object val;
        try {
            while (count == 0){
                condition.await();
            }
            takeIndex = takeIndex % maxLength;
            val = value[takeIndex++];
            count--;
            condition.signal();
        }finally {
           lock.unlock();
        }
        return (T) val;
    }
}

Answer 1

类似的问题:Why should wait() always be called inside a loop

解释

考虑这种情况:

1. 消费者 1 在 lock.lock(); 上被阻塞
2. 消费者 2 在 empty.await(); 上被阻止。
3. producer 持有锁并向队列中添加一个元素，这使得 count = 1 并调用 empty.signal();。
4. consumer 2收到这个信号，从empty.await();中唤醒，需要重新获取锁，而cosumer 1 领先于它。
5. cosumer 1 获得锁并发现计数为 1，因此它将计数减为 0。

6. cosumer 2 获得锁，因为它已经执行了

   if (count == 0){    <--- consumer 2 will not re-check this condition
       empty.await();  
   }
   

   cosumer 2 认为队列不为空，然后执行:

   takeIndex = takeIndex % maxLength;
   val = value[takeIndex++];
   count--;
   

   这使得计数递减为 0。

解决方案

使用 while 而不是 if 保证消费者 2 将重新检查队列是否为空，这保证 count >= 0。

while (count == 0){
    empty.await();
}

另外，最好用 produce 方法做同样的事情:

while (count.equals(maxLength)){
    log.info("The queue is full!");
    full.await();
}