I just want to achieve a blocking queue with ReentrantLock, I define two conditions full
and empty
, source code as follows:
@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;
}
}
When testing in two consumer threads and one provider thread, the count
is less than zero in some accidental time.
Why the blocking queue is not thread safe, who can help me, giving me some guidance? Thank you very mach!
Update(2018/10/17)
If I just use one Condition
, could it run correctly? Source code as follows:
@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;
}
}