JUC之阻塞队列

文章预览

阻塞队列

概念

阻塞队列与普通队列的区别在于，当队列是空的时，从队列中获取元素的操作将会被阻塞，或者当队列是满时，往队列里添加元素的操作会被阻塞。

试图从空的阻塞队列中获取元素的线程将会被阻塞，直到其他的线程往空的队列插入新的元素。同样，试图往已满的阻塞队列中添加新元素的线程同样也会被阻塞，直到其他的线程使队列重新变得空闲起来，如从队列中移除一个或者多个元素，或者完全清空队列。

下图展示了如何通过阻塞队列来合作：

线程1往阻塞队列中添加元素，而线程2从阻塞队列中移除元素

意义

在多线程领域:所谓阻塞,在某些情况下会挂起 线程(即线程阻塞),一旦条件满足,被挂起的线程优惠被自动唤醒

为什么需要使用BlockingQueue？

好处是我们不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为BlockingQueue都一手给你包办好了。在concurrent包 发布以前,在多线程环境下,我们每个程序员都必须自己去控制这些细节,尤其还要兼顾效率和线程安全, 而这会给我们的程序带来不小的复杂度.

Java的阻塞队列

• <mark>ArrayBlockingQueue: 由数组结构组成的有界阻塞队列</mark>
• <mark>LinkedBlockingDeque: 由链表结构组成的有界(但大小默认值为Integer.MAX_VALUE)阻塞队列.</mark>
• PriorityBlockingQueue:支持优先级排序的无界阻塞队列.
• DelayQueue: 使用优先级队列实现的延迟无界阻塞队列.
• <mark>SynchronousQueue:不存储元素的阻塞队列,也即是单个元素的队列.</mark>
• LinkedTransferQueue:由链表结构组成的无界阻塞队列.
• LinkedBlockingDeque:由了解结构组成的双向阻塞队列.

阻塞队列的核心方法

	抛出异常	返回特殊值	一直阻塞	超时退出
插入方法	add(o)	offer(o)	put(o)	offer(o,timeout, timeunit)
移除方法	remove(o)	poll()	take(o)	poll(o,timeout, timeunit)
检查方法	element()	peek()	—	—

方法效果	成功/队满	失败/队空
抛出异常	当阻塞队列满时，再往队列里add插入元素会抛IllegalStateException:Queue full	当阻塞队列空时，再往队列里remove移除元素会抛NoSuchElementException
返回特殊值	插入方法，成功ture失败false	移除方法，成功返回出队列的元素，队列里没有就返回null
一直阻塞	当阻塞队列满时，生产者线程继续往队列里put元素，队列会一直阻塞生产者线程直到put数据or响应中断退出	当阻塞队列空时，消费者线程试图从队列里take元素，队列会一直阻塞消费者线程直到队列可用
超时退出		当阻塞队列满时，队列会阻塞生产者线程一定时间，超过限时后生产者线程会退出

抛出异常

使用 add() 向阻塞队列添加数据，成功返回true。

当向队列满的时候抛出异常 java.lang.IllegalStateException: Queue full 。

使用 remove() 方法移除阻塞队列中的元素，从队列头先开始移除，移除成功后返回。

当队列为空时进行移除操作，抛出 java.util.NoSuchElementException 异常

使用 element() 返回当前阻塞队列的头元素。

当阻塞队列为空的时候使用抛出 java.util.NoSuchElementException 异常

/**
 * 阻塞队列
 */
public class BlockingQueueDemo {

    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        System.out.println(blockingQueue.add("a"));
        System.out.println(blockingQueue.add("b"));
        System.out.println(blockingQueue.add("c"));
        //java.lang.IllegalStateException: Queue full
        //System.out.println(blockingQueue.add("d"));
        //返回当前队列头元素
        System.out.println(blockingQueue.element());

        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        System.out.println(blockingQueue.remove());
        //java.util.NoSuchElementException
        //System.out.println(blockingQueue.remove());
        //java.util.NoSuchElementException
        System.out.println(blockingQueue.element());
    }
}

返回特殊值

使用 offer() 方法向阻塞队列中添加值，添加成功返回 true

当队列满时继续添加，返回 false 而不是抛出异常

使用 poll() 方法向阻塞队列中移除元素，并返回。

队列为空时移除，返回 null

使用 peek() 方法查看当前队列头元素。队列为空时返回 null

public class BlockingQueueDemo {
    
    public static void main(String[] args) {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        System.out.println(blockingQueue.offer("a"));
        System.out.println(blockingQueue.offer("b"));
        System.out.println(blockingQueue.offer("c"));
        //队列满
        System.out.println(blockingQueue.offer("d"));//false
        //查看当前队列头元素
        System.out.println(blockingQueue.peek());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        //队列空
        System.out.println(blockingQueue.poll());//null
        System.out.println(blockingQueue.peek());//null
    }
}

一直阻塞

使用 put() 方法向阻塞队列中添加元素，添加成功没有返回值，当队列满时，一直阻塞直到添加成功或者被中断才能停止。

使用 take() 方法向阻塞队列中移除元素，移除成功返回值，当队列为空时，一直阻塞直到再次移除成功或者被中断才能停止。

public class BlockingQueueDemo {
    
    public static void main(String[] args) throws InterruptedException {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        System.out.println("全部添加完成！");
        //队列满
        //blockingQueue.put("d");//false

        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        //队列为空
        System.out.println(blockingQueue.take());
    }
    

超时退出

添加使用的是 offer 的重载方法 offer(E e, long timeout, TimeUnit unit)

移除使用的是poll的重载方法 poll(E e, long timeout, TimeUnit unit)

两者都是在队列满，队列空是添加，移除元素等待给定的时间，如果在给定的时间中添加成功返回ture失败返回false，移除成功则返回元素，失败返回null。

public class BlockingQueueDemo {

    public static void main(String[] args) throws InterruptedException {

        BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
        System.out.println("全部添加完成！");
        //队列满 等待一秒
        System.out.println(blockingQueue.offer("d", 1l, TimeUnit.SECONDS));

        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        //队列为空 等待一秒
        System.out.println(blockingQueue.poll(1l, TimeUnit.SECONDS));
    }

}

SynchronousQueue

SynchronousQueue没有容量，与其他BlockingQueue不同，SynchronousQueue是一个不存储元素的BlockingQueue。每一个put操作必须要等待一个take操作，否则不能继续添加元素，反之亦然。

只有B把元素取走，A才能继续向队列终添加元素，否则一直阻塞。

public class SynchronousQueueDemo {

    public static void main(String[] args) {
        BlockingQueue<String> blockingQueue = new SynchronousQueue<>();

        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName()+"\t put1");
                blockingQueue.put("1");
                System.out.println(Thread.currentThread().getName()+"\t put2");
                blockingQueue.put("2");
                System.out.println(Thread.currentThread().getName()+"\t put3");
                blockingQueue.put("3");

            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"AA").start();

        new Thread(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
                TimeUnit.SECONDS.sleep(2);
                System.out.println(Thread.currentThread().getName()+"\t "+blockingQueue.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"BB").start();
    }
}

阻塞队列版生产者消费者

/**
 * volatile/CAS/atomicInteger/BlockQueue/线程交互/原子引用
 */
public class ProdConsumer_BlockQueueDemo {

    public static void main(String[] args) {
        MyResource myResource = new MyResource(new ArrayBlockingQueue<>(10));
        new Thread(() -> {
            System.out.println(Thread.currentThread().getName()+"\t 生产线程启动");
            try {
                myResource.myProd();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"Prod").start();

        new Thread(() -> {
            System.out.println(Thread.currentThread().getName()+"\t 消费线程启动");
            try {
                myResource.myConsumer();
            } catch (Exception e) {
                e.printStackTrace();
            }
        },"Consumer").start();

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println();
        System.out.println();
        System.out.println("5秒钟到，大老板叫停！");
        try {
            myResource.stop();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

class MyResource {
    private volatile boolean FLAG = true;//默认开启，进行生产+消费
    private AtomicInteger atomicInteger = new AtomicInteger();
    BlockingQueue<String> blockingQueue = null;

    public MyResource(BlockingQueue<String> blockingQueue) {
        this.blockingQueue = blockingQueue;
        System.out.println(blockingQueue.getClass().getName());
    }

    public void myProd() throws Exception {
        String data = null;
        boolean retValue;
        while (FLAG) {
            data = atomicInteger.incrementAndGet() + "";
            retValue = blockingQueue.offer(data, 2l, TimeUnit.SECONDS);
            if (retValue) {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列" + data + "成功");
            } else {
                System.out.println(Thread.currentThread().getName() + "\t 插入队列" + data + "失败");
            }
            TimeUnit.SECONDS.sleep(1);
        }
        System.out.println(Thread.currentThread().getName()+"\t大老板叫停了，表示FLAG=false，生产动作结束");

    }
    public void myConsumer() throws Exception{
        String result = null;
        while (FLAG){
            result = blockingQueue.poll(2L,TimeUnit.SECONDS);
            if (null==result||result.equalsIgnoreCase("")){
                System.out.println(Thread.currentThread().getName()+"\t 超过2秒钟没有取到蛋糕，消费退出");
                System.out.println();
                System.out.println();
                return;
            }
            System.out.println(Thread.currentThread().getName()+"\t 消费队列蛋糕"+result+"成功");
        }
    }
    public void stop() throws Exception{
        this.FLAG = false;
    }
}

………………………………