Java併發（四）BlockingQueue的使用

時間 2019-11-17

原文原文鏈接

wait()和notifyAll()方法以一種很是低級的方式解決了任務互操做的問題，即每次交互時都須要握手。在許多狀況下，你能夠瞄準更高的抽象級別，使用同步隊列來解決任務協做的問題。同步隊列在任什麼時候刻都只容許一個任務插入或移除元素。在java.util.concurrent.BlockingQueue接口中提供了這個隊列，這個接口有大量的標準實現。你一般可使用LinkedBlockingQueue，它是一個無界隊列，你還可使用ArrayBlockingQueue，它具備固定的尺寸，所以你能夠在它被阻塞以前，向其中放置有限數量的元素。java

若是消費者任務試圖從隊列中獲取對象，而該隊列此時爲空，那麼這些隊列還能夠掛起消費者任務，而且當有更多的元素可用時回覆消費者任務。阻塞隊列能夠解決很是大的問題，而其方式與wait()和notifyAll()相比，則要簡單並可靠許多。dom

考慮下面這個BlockingQueue的示例，有一臺機器具備三個任務：一個製做吐司，一個給吐司抹黃油，還有一個給吐司塗果醬。咱們能夠經過各個處理過程之間的BlockingQueue來運行這個吐司製做程序：ide

import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

class Toast {
    /**
     * 吐司的狀態：
     * DRY: 烘乾的
     * BUTTERED: 塗了黃油的
     * JAMMED: 塗了果醬的
     * <p>吐司的狀態只能由DRY->BUTTERED->JAMMED轉變
     */
    public enum Status {DRY, BUTTERED, JAMMED}
    private Status status = Status.DRY;//默認狀態爲DRY
    private final int id;
    public Toast(int id) { this.id =  id;}
    public void butter() {status = Status.BUTTERED;}
    public void jam() {status = Status.JAMMED;}
    public Status getStatus() {return status;}
    public int getId() {return id;}
    public String toString() {
        return "Toast id: " + id + ", status: " + status;
    }
}

@SuppressWarnings("serial")
class ToastQueue extends LinkedBlockingQueue<Toast> {}

/**
 * 生產吐司的任務。
 */
class Toaster implements Runnable {
    private ToastQueue toastQueue;
    private int count = 0;
    private Random random = new Random(47);
    public Toaster(ToastQueue queue) {
        this.toastQueue = queue;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                TimeUnit.MILLISECONDS.sleep(300 + random.nextInt(500));
                //生產一片吐司，這些吐司是有序的
                Toast toast = new Toast(count++);
                System.out.println(toast);
                //放到toastQueue中
                toastQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Toaster interrupted.");
        }
        System.out.println("Toaster off.");
    }
}

/**
 * 塗黃油的任務。
 */
class Butterer implements Runnable {
    private ToastQueue dryQueue;
    private ToastQueue butteredQueue;
    public Butterer(ToastQueue dryQueue, ToastQueue butteredQueue) {
        this.dryQueue = dryQueue;
        this.butteredQueue = butteredQueue;
    }
    
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一個吐司以前會一直阻塞
                Toast toast = dryQueue.take();
                toast.butter();
                System.out.println(toast);
                butteredQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Butterer interrupted.");
        }
        System.out.println("Butterer off.");
        
    }
}

/**
 * 塗果醬的任務。
 */
class Jammer implements Runnable {
    private ToastQueue butteredQueue;
    private ToastQueue finishedQueue;
    public Jammer(ToastQueue butteredQueue, ToastQueue finishedQueue) {
        this.finishedQueue = finishedQueue;
        this.butteredQueue = butteredQueue;
    }
    
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一個吐司以前會一直阻塞
                Toast toast = butteredQueue.take();
                toast.jam();
                System.out.println(toast);
                finishedQueue.put(toast);
            }
        } catch (InterruptedException e) {
            System.out.println("Jammer interrupted.");
        }
        System.out.println("Jammer off.");
    }
}

/**
 * 吃吐司的人，消費者。
 */
class Eater implements Runnable {
    private ToastQueue finishedQueue;
    private int count = 0;
    public Eater (ToastQueue finishedQueue) {
        this.finishedQueue = finishedQueue;
    }
    @Override
    public void run() {
        try {
            while(!Thread.interrupted()) {
                //在取得下一個吐司以前會一直阻塞
                Toast toast = finishedQueue.take();
                //驗證取得的吐司是有序的，並且狀態是JAMMED的
                if (toast.getId() != count++ || 
                        toast.getStatus() != Toast.Status.JAMMED) {
                    System.out.println("Error -> " + toast);
                    System.exit(-1);
                } else {
                    //吃掉吐司
                    System.out.println(toast + "->Eaten");
                }
            }
        } catch (InterruptedException e) {
            System.out.println("Eater interrupted.");
        }
        System.out.println("Eater off.");
    }
}

public class ToastOMatic {
    public static void main(String[] args) throws Exception {
        ToastQueue dryQueue = new ToastQueue();
        ToastQueue butteredQueue = new ToastQueue();
        ToastQueue finishedQueue = new ToastQueue();
        ExecutorService exec = Executors.newCachedThreadPool();
        exec.execute(new Toaster(dryQueue));
        exec.execute(new Butterer(dryQueue, butteredQueue));
        exec.execute(new Jammer(butteredQueue, finishedQueue));
        exec.execute(new Eater(finishedQueue));
        TimeUnit.SECONDS.sleep(5);
        exec.shutdownNow();
    }
}

執行結果（可能的結果）：this

Toast id: 0, status: DRY
Toast id: 0, status: BUTTERED
Toast id: 0, status: JAMMED
Toast id: 0, status: JAMMED->Eaten
Toast id: 1, status: DRY
Toast id: 1, status: BUTTERED
Toast id: 1, status: JAMMED
Toast id: 1, status: JAMMED->Eaten
Toast id: 2, status: DRY
Toast id: 2, status: BUTTERED
Toast id: 2, status: JAMMED
Toast id: 2, status: JAMMED->Eaten
Toast id: 3, status: DRY
Toast id: 3, status: BUTTERED
Toast id: 3, status: JAMMED
Toast id: 3, status: JAMMED->Eaten
Toast id: 4, status: DRY
Toast id: 4, status: BUTTERED
Toast id: 4, status: JAMMED
Toast id: 4, status: JAMMED->Eaten
Toast id: 5, status: DRY
Toast id: 5, status: BUTTERED
Toast id: 5, status: JAMMED
Toast id: 5, status: JAMMED->Eaten
Toast id: 6, status: DRY
Toast id: 6, status: BUTTERED
Toast id: 6, status: JAMMED
Toast id: 6, status: JAMMED->Eaten
Toast id: 7, status: DRY
Toast id: 7, status: BUTTERED
Toast id: 7, status: JAMMED
Toast id: 7, status: JAMMED->Eaten
Eater interrupted.
Eater off.
Butterer interrupted.
Toaster interrupted.
Toaster off.
Jammer interrupted.
Jammer off.
Butterer off.

Toast是一個使用enum值的優秀示例。注意，這個示例中沒有任何顯式的同步（即便用Lock對象或者synchronized關鍵字的同步），由於同步已經由隊列和系統的設計隱式的管理了——每片Toast在任什麼時候刻都只由一個任務在操做。由於隊列的阻塞，使得處理過程將被自動的掛起和恢復。你能夠看到由BlockingQueue產生的簡化十分明顯。在使用顯式的wait()和notifyAll()時存在的類和類之間的耦合被消除了，由於每一個類都只和它的BlockingQueue通訊。設計