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

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

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();
    }
    //www.cdtarena.com
    public synchronized void offer(Object object) throws InterruptedException{
        notifyAll();
        if(linkedList.size() == maxLength){
            wait();
        }
        linkedList.add(object);
    }
}

下面是测试代码：

[java] view plaincopyprint?

import java.util.concurrent.atomic.AtomicInteger;

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

                }
            }
        };
        //www.cdtarena.com 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();
        }
    }
}