阻塞队列

ArrayBlockingQueue

LinkedBlockingQueue

LinkedBlockingDQueue

...

操作方法

插入

add/offer()/put

获取(删除)

remove/poll/take

简单demo

public class BlockDemo {

ArrayBlockingQueue<String> arrayBlockingQueue = new ArrayBlockingQueue(10);

    {
        init();
    }

    private void init(){
        new Thread(()->{
            while (true){
                try {
                    String data = arrayBlockingQueue.take();
                    System.out.println("receive:"+data);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    public void add(String data) throws Exception{
        arrayBlockingQueue.add(data);
        System.out.println("sendData:"+data);
        Thread.sleep(1000L);
    }

    public static void main(String[] args){
        BlockDemo demo = new BlockDemo();
        for (int i = 0; i < 1000; i++) {
            try {
                demo.add("data:"+i);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

ArrayBlockingQueue内部实现

• 采用Object[] 进行数据存储
• add()时依次添加至下标0~(length-1) '游标'
• 数组填满后通过Condition 进程阻塞
• 通过ReentrantLock进行线程安全控制
• take()时 如果数组为空进行阻塞(直至有新的元素添加)

原子操作 Atomic

• 基于CAS实现

    public final int getAndAdd(int delta) {
        return unsafe.getAndAddInt(this, valueOffset, delta);
    }

    public final int getAndAddInt(Object var1, long var2, int var4) {
        int var5;
        do {
            var5 = this.getIntVolatile(var1, var2);
        } while(!this.compareAndSwapInt(var1, var2, var5, var5 + var4));

        return var5;
    }