一段阻塞队列代码的纠错与优化

下面的代码在某处发现后，立马发现存在问题。

public class BlockingQ {

private Object notEmpty = new Object();
private Object notFull = new Object();
private Queue<Object> linkedList = new LinkedList<Object>();
private int maxLength = 10;

public Object take() throws InterruptedException{
synchronized(notEmpty){
if(linkedList.size() == 0){
notEmpty.wait();
}
synchronized(notFull){
if(linkedList.size() == maxLength){
notFull.notifyAll();
}
return linkedList.poll();
}
}
}

public void offer(Object object) throws InterruptedException{
synchronized(notEmpty){
if(linkedList.size() == 0){
notEmpty.notifyAll();
}
synchronized(notFull){
if(linkedList.size() == maxLength){
notFull.wait();
}
linkedList.add(object);
}
}
}
}

最简单的情况就是一个生产者，两个消费者，

if(linkedList.size() == 0){
notEmpty.wait();
}

上面的if语句在第二线程重新获取锁后，进来的时候不会再次判断是否有消费元素，会直接返回null。

public Object take() throws InterruptedException{
synchronized(notEmpty){
while(linkedList.size() == 0){
notEmpty.wait();
}
synchronized(notFull){
if(linkedList.size() == maxLength){
notFull.notifyAll();
}
return linkedList.poll();
}
}
}

修正代码如上面所示。

整个代码主要用作示例，所以代码上存在很冗余的点，下面是代码优化后的版本：

public class SimpleBlockingQ {
private Queue<Object> linkedList = new LinkedList<Object>();
private int maxLength = 10;

public synchronized Object take() throws InterruptedException{
while(linkedList.size() == 0){
wait();
}
notifyAll();
return linkedList.poll();
}

public synchronized void offer(Object object) throws InterruptedException{
notifyAll();
if(linkedList.size() == maxLength){
wait();
}
linkedList.add(object);
}
}

从性能上优化，可以参考《一种简单无锁队列的实现》和《无锁同步栈实现》。

下面是测试代码：

import java.util.concurrent.atomic.AtomicInteger;

/**
* @author 天水
* @date 2013-4-12 下午03:44:15
*/
public class BlockingQTest {

public static AtomicInteger index = new AtomicInteger(0);

public static void main(String[] args){
int tCount = 10; // thread count

final BlockingQ BQ = new BlockingQ();
final SimpleBlockingQ SBQ = new SimpleBlockingQ();

// provider
Runnable pr = new Runnable(){
@Override
public void run() {
while(true){
try {
Thread.sleep(100);

int tindex = index.getAndIncrement();
//BQ.offer(tindex);
//System.out.println("BQ offer: " + tindex);
SBQ.offer(tindex);
System.out.println("SBQ offer: " + tindex);

Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}

}
}
};
// consumer
Runnable cr = new Runnable(){
@Override
public void run() {
while(true){
try {
Thread.sleep(100);

//System.out.println("BQ take: " + BQ.take());
System.out.println("SBQ take: " + SBQ.take());

Thread.sleep(100);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
};

for(int i=0; i<tCount; i++){
new Thread(cr).start();
}

for(int i=0; i<tCount; i++){
new Thread(pr).start();
}
}
}