Java多線程進(jìn)階（三一）—— J.U.C之collections框架：BlockingQueue接

entner 發(fā)布于2019-08-19 11:06 / 517人閱讀

摘要：和方法會(huì)一直阻塞調(diào)用線程，直到線程被中斷或隊(duì)列狀態(tài)可用和方法會(huì)限時(shí)阻塞調(diào)用線程，直到超時(shí)或線程被中斷或隊(duì)列狀態(tài)可用。

本文首發(fā)于一世流云專欄：https://segmentfault.com/blog...

一、引言

從本節(jié)開始，我們將介紹juc-collections框架中的“阻塞隊(duì)列”部分。阻塞隊(duì)列在實(shí)際應(yīng)用中非常廣泛，許多消息中間件中定義的隊(duì)列，通常就是一種“阻塞隊(duì)列”。

那么“阻塞隊(duì)列”和我們之前討論過(guò)的ConcurrentLinkedQueue、ConcurrentLinkedDeque有什么不同呢？

ConcurrentLinkedQueue和ConcurrentLinkedDeque是以非阻塞算法實(shí)現(xiàn)的高性能隊(duì)列，其使用場(chǎng)景一般是在并發(fā)環(huán)境下，需要“隊(duì)列”/“棧”這類數(shù)據(jù)結(jié)構(gòu)時(shí)才會(huì)使用；而“阻塞隊(duì)列”通常利用了“鎖”來(lái)實(shí)現(xiàn)，也就是會(huì)阻塞調(diào)用線程，其使用場(chǎng)景一般是在“生產(chǎn)者-消費(fèi)者”模式中，用于線程之間的數(shù)據(jù)交換或系統(tǒng)解耦。

在Java多線程基礎(chǔ)（七）——Producer-Consumer模式中，我們?cè)?jiǎn)要的談到過(guò)“生產(chǎn)者-消費(fèi)者”這種模式。在這種模式中，“生產(chǎn)者”和“消費(fèi)者”是相互獨(dú)立的，兩者之間的通信需要依靠一個(gè)隊(duì)列。這個(gè)隊(duì)列，其實(shí)就是本文中的“阻塞隊(duì)列”。

引入“阻塞隊(duì)列”的最大好處就是解耦，在軟件工程中，“高內(nèi)聚，低耦合”是進(jìn)行模塊設(shè)計(jì)的準(zhǔn)則之一，這樣“生產(chǎn)者”和“消費(fèi)者”其實(shí)是互不影響的，將來(lái)任意一方需要升級(jí)時(shí)，可以保證系統(tǒng)的平滑過(guò)渡。

二、BlockingQueue簡(jiǎn)介

BlockingQueue是在JDK1.5時(shí)，隨著J.U.C引入的一個(gè)接口：

BlockingQueue繼承了Queue接口，提供了一些阻塞方法，主要作用如下：

當(dāng)線程向隊(duì)列中插入元素時(shí)，如果隊(duì)列已滿，則阻塞線程，直到隊(duì)列有空閑位置（非滿）；

當(dāng)線程從隊(duì)列中取元素（刪除隊(duì)列元素）時(shí)，如果隊(duì)列未空，則阻塞線程，直到隊(duì)列有元素；

既然BlockingQueue是一種隊(duì)列，所以也具備隊(duì)列的三種基本方法：插入、刪除、讀取：

操作類型	拋出異常	返回特殊值	阻塞線程	超時(shí)
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
刪除	remove()	poll()	take()	poll(time, unit)
讀取	element()	peek()	/	/

可以看到，對(duì)于每種基本方法，“拋出異常”和“返回特殊值”的方法定義和Queue是完全一樣的。BlockingQueue只是增加了兩類和阻塞相關(guān)的方法：put(e)、take()；offer(e, time, unit)、poll(time, unit)。

put(e)和take()方法會(huì)一直阻塞調(diào)用線程，直到線程被中斷或隊(duì)列狀態(tài)可用；
offer(e, time, unit)和poll(time, unit)方法會(huì)限時(shí)阻塞調(diào)用線程，直到超時(shí)或線程被中斷或隊(duì)列狀態(tài)可用。

public interface BlockingQueue extends Queue {
    /**
     * 插入元素e至隊(duì)尾, 如果隊(duì)列已滿, 則阻塞調(diào)用線程直到隊(duì)列有空閑空間.
     */
    void put(E e) throws InterruptedException;

    /**
     * 插入元素e至隊(duì)列, 如果隊(duì)列已滿, 則限時(shí)阻塞調(diào)用線程，直到隊(duì)列有空閑空間或超時(shí).
     */
    boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException;

    /**
     * 從隊(duì)首刪除元素，如果隊(duì)列為空, 則阻塞調(diào)用線程直到隊(duì)列中有元素.
     */
    E take() throws InterruptedException;

    /**
     * 從隊(duì)首刪除元素，如果隊(duì)列為空, 則限時(shí)阻塞調(diào)用線程，直到隊(duì)列中有元素或超時(shí).
     */
    E poll(long timeout, TimeUnit unit) throws InterruptedException;

    // ...
}

除此之外，BlockingQueue還具有以下特點(diǎn)：

BlockingQueue隊(duì)列中不能包含null元素；

BlockingQueue接口的實(shí)現(xiàn)類都必須是線程安全的，實(shí)現(xiàn)類一般通過(guò)“鎖”保證線程安全；

BlockingQueue 可以是限定容量的。remainingCapacity()方法用于返回剩余可用容量，對(duì)于沒(méi)有容量限制的BlockingQueue實(shí)現(xiàn)，該方法總是返回Integer.MAX_VALUE 。

三、再談“生產(chǎn)者-消費(fèi)者”模式

最后，我們來(lái)看下如何利用BlockingQueue來(lái)實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模式。在生產(chǎn)者-消費(fèi)者模式中，一共有四類角色：生產(chǎn)者、消費(fèi)者、消息隊(duì)列、消息體。我們利用BlockingQueue來(lái)實(shí)現(xiàn)消息隊(duì)列，其余部分沒(méi)有什么變化。

Producer(生產(chǎn)者)

生產(chǎn)者生產(chǎn)消息體（Data），并將消息體（Data）傳遞給通道（Channel）。

/**
 * 生產(chǎn)者
 */
public class Producer implements Runnable {

    private Channel channel;

    public Producer(Channel channel) {
        this.channel = channel;
    }

    @Override
    public void run() {

        while (true) {
            String v = String.valueOf(ThreadLocalRandom.current().nextInt());
            Data data = new Data(v);
            try {
                channel.put(data);
                System.out.println(Thread.currentThread().getName() + " produce :" + data);

            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            Thread.yield();
        }
    }
}

Consumer(消費(fèi)者)

消費(fèi)者從通道（Channel）中獲取數(shù)據(jù)，進(jìn)行處理。

/**
 * 消費(fèi)者
 */
public class Consumer implements Runnable {
    private final Channel channel;

    public Consumer(Channel channel) {
        this.channel = channel;
    }

    @Override
    public void run() {
        while (true) {

            try {
                Object obj = channel.take();
                System.out.println(Thread.currentThread().getName() + " consume :" + obj.toString());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            Thread.yield();
        }
    }
}

Channel（通道）

相當(dāng)于消息的隊(duì)列，對(duì)消息進(jìn)行排隊(duì)，控制消息的傳輸。

/**
 * 通道類
 */
public class Channel {

    private final BlockingQueue blockingQueue;


    public Channel(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    public Object take() throws InterruptedException {
        return blockingQueue.take();
    }

    public void put(Object o) throws InterruptedException {
        blockingQueue.put(o);
    }

}

Data（消息體/數(shù)據(jù)）

Data代表了實(shí)際生產(chǎn)或消費(fèi)的數(shù)據(jù)。

/**
 * 數(shù)據(jù)/消息
 */
public class Data implements Serializable {
    private T data;

    public Data(T data) {
        this.data = data;
    }

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }

    @Override
    public String toString() {
        return "Data{" +
            "data=" + data +
            "}";
    }
}

調(diào)用如下：

public class Main {
    public static void main(String[] args) {
        BlockingQueue blockingQueue = new SomeQueueImplementation();
        Channel channel = new Channel(blockingQueue);

        Producer p = new Producer(channel);
        Consumer c1 = new Consumer(channel);
        Consumer c2 = new Consumer(channel);

        new Thread(p).start();
        new Thread(c1).start();
        new Thread(c2).start();

    }
}