JUC-BlockingQueue

阻塞队列BlockingQueue

四组API

public class BlockingQueueTest {

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

    /**
     * 抛出异常
     * add()
     * remove()
     */
    public static void test1() {
        // 设置阻塞队列大小为3
        BlockingQueue<Object> blockingQueue = new ArrayBlockingQueue<>(3);

        blockingQueue.add("a");
        blockingQueue.add("b");
        blockingQueue.add("c");
        // java.lang.IllegalStateException: Queue full，抛异常
        //blockingQueue.add("d");

        blockingQueue.remove();
        blockingQueue.remove();
        blockingQueue.remove();
        // java.util.NoSuchElementException，抛异常
        blockingQueue.remove();
    }

    /**
     * 不抛出异常，有返回值 true 或 false
     * offer()
     * poll()
     * @throws InterruptedException
     */
    public static void test2() throws InterruptedException {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

        blockingQueue.offer("a");
        blockingQueue.offer("b");
        blockingQueue.offer("c");
        System.out.println(blockingQueue.offer("d")); // false

        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
        System.out.println(blockingQueue.poll());
    }

    /**
     * 阻塞一直等待
     * put()
     * take()
     * @throws InterruptedException
     */
    public static void test3() throws InterruptedException {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

        blockingQueue.put("a");
        blockingQueue.put("b");
        blockingQueue.put("c");
//        blockingQueue.put("d"); // 无返回值，一直阻塞，等待

        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take());
        System.out.println(blockingQueue.take()); // 阻塞，一直等待
    }

    /**
     * 超时等待
     * offer(,,)
     * poll(,)
     * @throws InterruptedException
     */
    public static void test4() throws InterruptedException {
        BlockingQueue blockingQueue = new ArrayBlockingQueue(3);

        blockingQueue.offer("a");
        blockingQueue.offer("b");
        blockingQueue.offer("d");
        blockingQueue.offer("d", 1, TimeUnit.SECONDS); // 阻塞，超时等待

        System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));
        System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));
        System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS));
        System.out.println(blockingQueue.poll(2, TimeUnit.SECONDS)); // null
    }
}

同步队列SynchronousQueue

public class SynchronousQueueTest {
    public static void main(String[] args) {
        // 同步队列，不会存储元素，put一个元素之后就会等待take
        // take一个元素就会等待put
        BlockingQueue<String> synchronousQueue = new SynchronousQueue<>();

        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName() + " put 1");
                synchronousQueue.put("1");
                System.out.println(Thread.currentThread().getName() + " put 2");
                synchronousQueue.put("2");
                System.out.println(Thread.currentThread().getName() + " put 3");
                synchronousQueue.put("3");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

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

内容来自狂神JUC课程