BLOKING

QUEUE

阻塞隊(duì)列，英文名BlockingQueue，顧名思義，首先它是一個(gè)隊(duì)列。阻塞就是說(shuō)在多線程環(huán)境下，線程在某些情況下會(huì)被掛起，這就是阻塞，一旦滿足條件，又會(huì)被喚醒。那么阻塞隊(duì)列的阻塞體現(xiàn)在何處？當(dāng)一個(gè)線程從阻塞隊(duì)列中取元素時(shí)，如果隊(duì)列為空了，那么取元素的操作就會(huì)被阻塞，直到有其他線程往隊(duì)列中添加了元素；當(dāng)一個(gè)線程往阻塞隊(duì)列中添加元素時(shí)，如果隊(duì)列滿了，那么添加元素的操作也會(huì)被阻塞，直到有其他線程從隊(duì)列中取走了元素。

有了阻塞隊(duì)列，我們不需要關(guān)心何時(shí)阻塞線程，何時(shí)喚醒線程。因?yàn)檫@些操作阻塞隊(duì)列都幫我們做了。隊(duì)列為空那么取元素的線程會(huì)自動(dòng)被阻塞，隊(duì)列已滿那么添加元素的線程會(huì)自動(dòng)阻塞。

之前我們只知道 Collection 下面有 set 和 list，其實(shí) queue也是繼承了 Collection，queue的子類就是 BlockingQueue，不過(guò)它還是接口，它總共有九個(gè)實(shí)現(xiàn)類，下面挑三個(gè)最重要的說(shuō)一說(shuō)。

• ArrayBlockingQueue：按照 ArrayList 來(lái)理解，由數(shù)組結(jié)構(gòu)組成的有界阻塞隊(duì)列。何為有界？其實(shí)就是隊(duì)列的容量?？匆幌滤臉?gòu)造方法：

1public ArrayBlockingQueue(int capacity) {
2        this(capacity, false);
3}

容量調(diào)用構(gòu)造方法時(shí)必傳的參數(shù)，默認(rèn)是非公平的，也就是說(shuō)，它的方法都用Lock上了鎖，我們知道lock可以通過(guò)true或者false來(lái)指定使用公平鎖還是非公平鎖。所以這里傳的false或者true其實(shí)是給lock用的。

• LinkedBlockingQueue：按照 LinkedList 來(lái)理解，由鏈表結(jié)構(gòu)組成的有界阻塞隊(duì)列。其實(shí)這里有個(gè)坑爹的地方，先來(lái)看一下它的構(gòu)造方法：

1public LinkedBlockingQueue() {
2        this(Integer.MAX_VALUE);
3}

它有個(gè)無(wú)參構(gòu)造，我們可以不傳容量，默認(rèn)是 ** int類型的最大值**，這不是坑爹嘛，int類型最大值是21億多，這不就相當(dāng)于無(wú)界嘛。當(dāng)然，它也有帶參構(gòu)造，可以指定容量，但是不能指定使用公平還是非公平鎖，默認(rèn)使用的是非公平鎖。

• SynchronousQueue：這個(gè)隊(duì)列比較特殊，只能存儲(chǔ)一個(gè)元素。里面有一個(gè)元素的時(shí)候，添加的線程就會(huì)被阻塞。

阻塞隊(duì)列的核心方法有四組，如下表：

1、用在哪？
阻塞隊(duì)列有哪些應(yīng)用呢？常見(jiàn)的有以下三個(gè)：

• 生產(chǎn)消費(fèi)模式；

• 線程池；

• 消息中間件。

本文將講解生產(chǎn)消費(fèi)模式中如何使用阻塞隊(duì)列。

2、生產(chǎn)消費(fèi)模式：

• 多線程編程的口訣：
  線程 操縱 資源類，判斷 干活 通知；防止虛假喚醒(判斷一定要用while，不能用if)。

• 基礎(chǔ)版生產(chǎn)消費(fèi)模式：
  資源類：

 1class Resource{
 2    private  Integer num = 0;
 3    private Lock lock = new ReentrantLock();
 4    private Condition condition = lock.newCondition();
 5    public void produce(){ // 生產(chǎn)的方法
 6        lock.lock();
 7        try {
 8            // 1.判斷
 9            while (num != 0){
10                // 等待，不能生產(chǎn)
11                condition.await();
12            }
13            // 2.干活
14            num ++;
15            System.out.println(Thread.currentThread().getName() + "\t" + num);
16            // 3.通知
17            condition.signalAll();
18        }catch (Exception e){
19            e.printStackTrace();
20        }finally {
21            lock.unlock();
22        }
23    }
24
25    public void consum(){ // 消費(fèi)的方法
26        lock.lock();
27        try {
28            // 1.判斷
29            while (num == 0){
30                // 等待，不能消費(fèi)
31                condition.await();
32            }
33            // 2.干活
34            num --;
35            System.out.println(Thread.currentThread().getName() + "\t" + num);
36            // 3.通知
37            condition.signalAll();
38        }catch (Exception e){
39            e.printStackTrace();
40        }finally {
41            lock.unlock();
42        }
43    }
44}

線程操縱資源類：

 1public static void main(String[] args){
 2        Resource resource = new Resource();
 3        new Thread(() -> {
 4            for (int i=1; i<=5; i++){
 5                resource.produce();
 6            }
 7        },"A").start();
 8
 9        new Thread(() -> {
10            for (int i=1; i<=5; i++){
11                resource.consum();
12            }
13        },"B").start();
14}

這就是基礎(chǔ)版的生產(chǎn)消費(fèi)模式。準(zhǔn)確的說(shuō)應(yīng)該是2.0版本，最開(kāi)始學(xué)的是使用synchronized實(shí)現(xiàn)的。那么，synchronized和lock到底有什么區(qū)別呢？

• synchronized 和 lock 的區(qū)別：

這里來(lái)說(shuō)一說(shuō)ReentrantLock的精確喚醒。現(xiàn)有題目如下：

1有A、B、C三個(gè)線程，A打印5次，B打印10次，C打印15次，然后又是A打印5次，B打印10次，C打印15次……循環(huán)10輪。

這就是經(jīng)典的線程按序交替問(wèn)題。看看如何使用 ReentrantLock 來(lái)解決。

 1/** 資源類 */
 2class Resource{
 3    private   Integer flag = 1; // 1: A執(zhí)行，2：B執(zhí)行，3：C執(zhí)行
 4    private  Lock lock = new ReentrantLock();
 5    public Condition condition1 = lock.newCondition();
 6    public Condition condition2 = lock.newCondition();
 7    public  Condition condition3 = lock.newCondition();
 8
 9    public void print(Condition waitCondition, Condition signalCondition, 
10                      Integer num, Integer nowFlag, Integer changedFlag){
11        lock.lock();
12        try { // 1. 判斷
13            while (this.flag != nowFlag){
14                waitCondition.await();
15            }
16            // 2. 干活
17            for (int i=1; i<=num; i++){
18                System.out.println(Thread.currentThread().getName() + "\t" + i);
19            }
20            // 3. 通知
21            this.flag = changedFlag;
22            signalCondition.signal();
23        }catch (Exception e){
24        }finally {
25            lock.unlock();
26        }
27    }
28}

這是資源類，一把鎖，因?yàn)橛腥齻€(gè)線程，需要精確喚醒，就需要三個(gè)condition；flag是一個(gè)標(biāo)識(shí)，用來(lái)判斷是哪個(gè)線程進(jìn)行執(zhí)行；print方法有5個(gè)參數(shù)，第一個(gè)是等待的線程的condition，第二個(gè)是需要喚醒的線程的condition，第三個(gè)是打印的次數(shù)，第四個(gè)是當(dāng)前的flag，第五個(gè)是需要修改的flag值。看看線程如何操作這個(gè)資源類：

 1public static void main(String[] args){
 2        Resource resource = new Resource();
 3        new Thread(() -> {
 4            for (int i=1; i<=10; i++){
 5                resource.print(resource.condition1, resource.condition2, 5, 1, 2);
 6                System.out.println("========================================================");
 7               }
 8           },"A").start();
 9
10        new Thread(() -> {
11            for (int i=1; i<=10; i++){
12                resource.print(resource.condition2, resource.condition3, 10, 2, 3);
13                System.out.println("========================================================");
14            }
15        },"B").start();
16        new Thread(() -> {
17            for (int i=1; i<=10; i++){
18                resource.print(resource.condition3, resource.condition1, 15, 3, 1);
19                System.out.println("========================================================");
20            }
21        },"C").start();
22}

首先創(chuàng)建A線程，看看A線程調(diào)用print方法傳入?yún)?shù)后是什么情況：

 1lock.lock();
 2try { // 1. 判斷
 3    while (this.flag != 1){
 4        condition1.await();
 5    }
 6    // 2. 干活
 7    for (int i=1; i<=5; i++){
 8        System.out.println(Thread.currentThread().getName() + "\t" + i);
 9    }
10    // 3. 通知
11    this.flag = 2;
12    condition2.signal();
13}catch (Exception e){ 
14}finally {
15    lock.unlock();
16}

線程A調(diào)用時(shí)，首先判斷flag是不是1，如果不是，那么線程A就等待，否則就干活，打印5次。干完活要讓線程B執(zhí)行，所以將flag修改為2，然后將線程B喚醒。線程B調(diào)用時(shí)，自己干完活就喚醒C，線程C干完活就喚醒A……所以執(zhí)行結(jié)果就是：

• 阻塞隊(duì)列版生產(chǎn)消費(fèi)模式：
  資源類：

 1class Resource {
 2    private volatile boolean flag = true; // 標(biāo)識(shí)
 3    private AtomicInteger atomicInteger = new AtomicInteger();
 4    BlockingQueue<Integer> blockingQueue = null;
 5
 6    public Resource(BlockingQueue<Integer> blockingQueue) {
 7        this.blockingQueue = blockingQueue;
 8    }
 9
10    // 生產(chǎn)的方法
11    public void produce() throws Exception {
12        Integer data = null;
13        boolean result;
14        while (flag) {
15            data = atomicInteger.incrementAndGet();
16            result = blockingQueue.offer(data, 2L, TimeUnit.SECONDS);
17            if (result)
18                System.out.println(Thread.currentThread().getName() +"成功生產(chǎn)"+ data +" 號(hào)蛋糕！");
19            else
20                System.out.println(Thread.currentThread().getName() + "生產(chǎn)蛋糕失??！");
21            TimeUnit.SECONDS.sleep(1);
22        }
23        System.out.println(Thread.currentThread().getName() + "停止生產(chǎn)！");
24    }
25
26    // 消費(fèi)的方法
27    public void consume() throws Exception {
28        System.out.println();
29        Integer data = null;
30       while (flag){
31           data = blockingQueue.poll(2L, TimeUnit.SECONDS);
32           if (data == null){
33               flag = false;
34               System.out.println(Thread.currentThread().getName()+"超過(guò)2秒沒(méi)取到，停止消費(fèi)！");
35               return;
36           }
37           System.out.println(Thread.currentThread().getName() +"成功消費(fèi)" +data+ " 號(hào)蛋糕！");
38       }
39    }
40
41    // 停止的方法
42    public void stop() throws Exception {
43        this.flag = false;
44    }
45}

首先，當(dāng)flag為true時(shí)，進(jìn)行生產(chǎn)，就是將 atomicInteger 進(jìn)行自增，放到阻塞隊(duì)列中；放完一個(gè)就休息1秒鐘。然后是消費(fèi)的方法，也是當(dāng)flag為true就進(jìn)行消費(fèi)，消費(fèi)就是從阻塞隊(duì)列中取出元素，如果取到的是 null，說(shuō)明隊(duì)列中沒(méi)有元素了，就將flaf設(shè)為false，退出循環(huán)，停止消費(fèi)。最后的停止的方法，就是將flag設(shè)為false，這樣生產(chǎn)和消費(fèi)都會(huì)停止。
操縱資源類：

 1public static void main(String[] args) throws Exception{
 2        Resource resource = new Resource(new ArrayBlockingQueue<>(6));
 3        new Thread(() -> {
 4            System.out.println("生產(chǎn)線程啟動(dòng)！");
 5            try {
 6                resource.produce();
 7            } catch (Exception e) {
 8                e.printStackTrace();
 9            }
10
11        }, "生產(chǎn)線程 ").start();
12
13        new Thread(() -> {
14            System.out.println("消費(fèi)線程啟動(dòng)！");
15            try {
16                resource.consume();
17            } catch (Exception e) {
18                e.printStackTrace();
19            }
20        }, "消費(fèi)線程 ").start();
21        // 5秒鐘后停止
22        TimeUnit.SECONDS.sleep(5);
23        System.out.println();
24        System.out.println("5秒鐘后停止生產(chǎn)！");
25        resource.stop();
26}

這里是創(chuàng)建了兩個(gè)線程，一個(gè)調(diào)用生產(chǎn)方法，一個(gè)調(diào)用消費(fèi)方法；然后讓主線程睡5秒調(diào)用停止的方法。看看運(yùn)行結(jié)果：

這就是阻塞隊(duì)列版的生產(chǎn)消費(fèi)模式，不用我們?nèi)タ刂凭€程的通信。