基础 | 并发编程 - [阻塞队列]

§1 什么是阻塞队列

阻塞队列的存取性质与普通队列一致，即先进先出
同时，阻塞队列提供具有下面特性的存取方法

• 若队列是空的，获取操作阻塞，直到其他线程向其中存入元素
• 若队列是满的，存入操作阻塞，直到其他线程向其中获取元素或清空队列

§2 阻塞队列有什么好处

阻塞队列可以在上述条件下自动阻塞（挂起）线程，
随后，在满足条件时自动唤醒线程
编写逻辑时，不用考虑线程相关的问题，而只需要专注于容器
原始多线程实现方式见 基础 | 并发编程 - [Lock 使用 & 对比 synchronized] 中示例

§3 阻塞队列总览

类型一览

特性说明

• 有界
  队列的长度是由限制的
• 无界
  队列的长度没有限制，可以扩充至资源耗尽
• 支持优先级
  队列中的元素具有优先级，从队列中获取元素时会得到优先级最高或最低的元素
• 双向
  队列不再是只能从一端存入从另一端获取
  而是两端都可以存取
  可以作为栈或队列使用

常用 api 一览
作为容器，阻塞队列的核心功能也是 存 和 取
但阻塞队列为存取提供了多个接口，并配合多种反馈效果

• 阻塞
  能存取时存取，不能存取时等待
  直到 有空位可以存 或 有元素可以取
  见 §1 什么是阻塞队列
• 超时
  能存取时存取，不能存取时先阻塞
  直到 有空位可以存 或 有元素可以取，或超时
  超时时返回标记值
• 标记值
  能存取时存取，不能存取时返回标记值（false / null）
• 抛异常
  能存取时存取，不能存取时抛异常

element() 和 peek() 会返回队列的第一个元素
但队列为空（empty）时，前者抛异常，后者返回 null

通用 api 示例
仅以 ArrayBlockingQueue 为例

    public static void main(String[] args) {
        BlockingQueue<String> queue = new ArrayBlockingQueue<>(1);
        try {
            // 阻塞卡死 ============================================
            queue.put("a");
            System.out.println("put 1 over"); // put 1 over
            queue.put("b"); // block here
            System.out.println("put 2 over"); //not output
            queue.take();
            System.out.println("take 1 over"); // take 1 over
            queue.take(); // block here
            System.out.println("take 2 over"); //not output

            // 阻塞后恢复 ============================================
            queue.put("c");
            System.out.println("put 3 over"); // put 3 over
            new Thread(()->{
                // block here, until thread named "BBB" is over
                try { queue.put("C"); } catch (InterruptedException e) { e.printStackTrace(); }
            },"AAA").start();
            new Thread(()->{
                // take the "c" out
                try { TimeUnit.SECONDS.sleep(1); queue.take(); } catch (InterruptedException e) { e.printStackTrace(); }
            },"BBB").start();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 先阻塞，超时后返回标记值 ============================================
        try {
            queue.offer("a",1L,TimeUnit.SECONDS); // true
            queue.offer("b",1L,TimeUnit.SECONDS); // 1 second later, false
            queue.poll(1L,TimeUnit.SECONDS); // a
            queue.poll(1L,TimeUnit.SECONDS); // 1 second later, null
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        // 标记值 ============================================
        queue.offer("a"); // true
        queue.offer("b"); // false
        queue.peek(); // first element
        queue.poll(); // a
        queue.poll(); // null
        queue.peek(); // null

        // 一言不合抛异常  ============================================
        queue.add("a"); // true
        queue.add("b"); // java.lang.IllegalStateException: Queue full
        queue.element(); // first element

        queue.remove(); // element
        queue.remove(); // java.util.NoSuchElementException
        queue.element(); // java.util.NoSuchElementException
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

§4 SynchronousQueue

SynchronousQueue 不存储元素，存入一个元素后，此元素不取出，不能存入另一个元素
Synchronous 的意思是指此队列会使一对存取操作同步，如果不能达成同步则会阻塞，直到等到另一个匹配的操作

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

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

    new Thread(()->{
        try { TimeUnit.SECONDS.sleep(2);queue.take(); } catch (InterruptedException e) { e.printStackTrace(); }
        System.out.println("1 ---->");
        try { TimeUnit.SECONDS.sleep(2);queue.take(); } catch (InterruptedException e) { e.printStackTrace(); }
        System.out.println("2 ---->");
        try { TimeUnit.SECONDS.sleep(2);queue.take(); } catch (InterruptedException e) { e.printStackTrace(); }
        System.out.println("3 ---->");
    }).start();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21