自拍偷在线精品自拍偷,亚洲欧美中文日韩v在线观看不卡

【并發(fā)編程技術(shù)】聊聊阻塞隊列那些事

作者：志哥聊技術(shù)
開發(fā) 前端
開發(fā)者可以通過BlockingQueue阻塞隊列接口，簡單的代碼編程即可實現(xiàn)多線程中數(shù)據(jù)高效安全傳輸?shù)哪康?，確切的說，它幫助開發(fā)者減輕了不少的編程難度。

01、背景介紹

在之前的文章中，我們介紹了生產(chǎn)者和消費者模型的最基本實現(xiàn)思路，相信大家對它已經(jīng)有一個初步的認(rèn)識。

在 Java 的并發(fā)包里面還有一個非常重要的接口：BlockingQueue。

BlockingQueue是一個阻塞隊列，更為準(zhǔn)確的解釋是：BlockingQueue是一個基于阻塞機制實現(xiàn)的線程安全的隊列。通過它也可以實現(xiàn)生產(chǎn)者和消費者模型，并且效率更高、安全可靠，相比之前介紹的生產(chǎn)者和消費者模型，它可以同時實現(xiàn)生產(chǎn)者和消費者并行運行。

圖片圖片

那什么是阻塞隊列呢？

簡單的說，就是當(dāng)參數(shù)在入隊和出隊時，通過加鎖的方式來避免線程并發(fā)操作時導(dǎo)致的數(shù)據(jù)異常問題。

在 Java 中，能對線程并發(fā)執(zhí)行進(jìn)行加鎖的方式主要有synchronized和ReentrantLock，其中BlockingQueue采用的是ReentrantLock方式實現(xiàn)。

與此對應(yīng)的還有非阻塞機制的隊列，主要是采用 CAS 方式來控制并發(fā)操作，例如：ConcurrentLinkedQueue，這個我們在后面的文章再進(jìn)行分享介紹。

今天我們主要介紹BlockingQueue相關(guān)的知識和用法，廢話不多說了，進(jìn)入正題！

02、BlockingQueue 方法介紹

打開BlockingQueue的源碼，你會發(fā)現(xiàn)它繼承自Queue，正如上文提到的，它本質(zhì)是一個隊列接口。

public interface BlockingQueue<E> extends Queue<E> {
 //...省略
}

關(guān)于隊列，我們在之前的集合系列文章中對此有過深入的介紹，本篇就再次簡單的介紹一下。

隊列其實是一個數(shù)據(jù)結(jié)構(gòu)，元素遵循先進(jìn)先出的原則，所有新元素的插入，也被稱為入隊操作，會插入到隊列的尾部；元素的移除，也被稱為出隊操作，會從隊列的頭部開始移除，從而保證先進(jìn)先出的原則。

在Queue接口中，總共有 6 個方法，可以分為 3 類，分別是：插入、移除、查詢，內(nèi)容如下：

方法

描述

add(e)

插入元素，如果插入失敗，就拋異常

offer(e)

插入元素，如果插入成功，就返回 true；反之 false

remove()

移除元素，如果移除失敗，就拋異常

poll()

移除元素，如果移除成功，返回 true；反之 false

element()

獲取隊首元素，如果獲取結(jié)果為空，就拋異常

peek()

獲取隊首元素，如果獲取結(jié)果為空，返回空對象

因為BlockingQueue是Queue的子接口，了解Queue接口里面的方法，有助于我們對BlockingQueue的理解。

除此之外，BlockingQueue還單獨擴展了一些特有的方法，內(nèi)容如下：

方法

描述

put(e)

插入元素，如果沒有插入成功，線程會一直阻塞，直到隊列中有空間再繼續(xù)

offer(e, time, unit)

插入元素，如果在指定的時間內(nèi)沒有插入成功，就返回 false；反之 true

take()

移除元素，如果沒有移除成功，線程會一直阻塞，直到隊列中新的數(shù)據(jù)被加入

poll(time, unit)

移除元素，如果在指定的時間內(nèi)沒有移除成功，就返回 false；反之 true

drainTo(Collection c, int maxElements)

一次性取走隊列中的數(shù)據(jù)到 c 中，可以指定取的個數(shù)。該方法可以提升獲取數(shù)據(jù)效率，不需要多次分批加鎖或釋放鎖

分析源碼，你會發(fā)現(xiàn)相比普通的Queue子類，BlockingQueue子類主要有以下幾個明顯的不同點：

  • 1.元素插入和移除時線程安全：主要是通過在入隊和出隊時進(jìn)行加鎖，保證了隊列線程安全，加鎖邏輯采用ReentrantLock實現(xiàn)
  • 2.支持阻塞的入隊和出隊方法：當(dāng)隊列滿時，會阻塞入隊的線程，直到隊列不滿；當(dāng)隊列為空時，會阻塞出隊的線程，直到隊列中有元素；同時支持超時機制，防止線程一直阻塞

03、BlockingQueue 用法詳解

打開源碼，BlockingQueue接口的實現(xiàn)類非常多，我們重點講解一下其中的 5 個非常重要的實現(xiàn)類，分別如下表所示。

實現(xiàn)類

功能

ArrayBlockingQueue

基于數(shù)組的阻塞隊列，使用數(shù)組存儲數(shù)據(jù)，需要指定長度，所以是一個有界隊列

LinkedBlockingQueue

基于鏈表的阻塞隊列，使用鏈表存儲數(shù)據(jù)，默認(rèn)是一個無界隊列；也可以通過構(gòu)造方法中的capacity設(shè)置最大元素數(shù)量，所以也可以作為有界隊列

SynchronousQueue

一種沒有緩沖的隊列，生產(chǎn)者產(chǎn)生的數(shù)據(jù)直接會被消費者獲取并且立刻消費

PriorityBlockingQueue

基于優(yōu)先級別的阻塞隊列，底層基于數(shù)組實現(xiàn)，是一個無界隊列

DelayQueue

延遲隊列，其中的元素只有到了其指定的延遲時間，才能夠從隊列中出隊

下面我們對以上實現(xiàn)類的用法，進(jìn)行一一介紹。

3.1、ArrayBlockingQueue

ArrayBlockingQueue是一個基于數(shù)組的阻塞隊列，初始化的時候必須指定隊列大小，源碼實現(xiàn)比較簡單，采用的是ReentrantLock和Condition實現(xiàn)生產(chǎn)者和消費者模型，部分核心源碼如下：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

 /** 使用數(shù)組存儲隊列中的元素 */
 final Object[] items;

 /** 使用獨占鎖ReetrantLock */
 final ReentrantLock lock;

 /** 等待出隊的條件 */
 private final Condition notEmpty;

 /** 等待入隊的條件 */
 private final Condition notFull;

 /** 初始化時，需要指定隊列大小 */
 public ArrayBlockingQueue(int capacity) {
        this(capacity, false);
    }

    /** 初始化時，也指出指定是否為公平鎖， */
    public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

    /**入隊操作*/
    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }

    /**出隊操作*/
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }
}

ArrayBlockingQueue采用ReentrantLock進(jìn)行加鎖，只有一個ReentrantLock對象，這意味著生產(chǎn)者和消費者無法并行運行。

我們看一個簡單的示例代碼如下：

public class Container {

    /**
     * 初始化阻塞隊列
     */
    private final BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);

    /**
     * 添加數(shù)據(jù)到阻塞隊列
     * @param value
     */
    public void add(Integer value) {
        try {
            queue.put(value);
            System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 從阻塞隊列獲取數(shù)據(jù)
     */
    public void get() {
        try {
            Integer value = queue.take();
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}
/**
 * 生產(chǎn)者
 */
public class Producer extends Thread {

    private Container container;

    public Producer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.add(i);
        }
    }
}
/**
 * 消費者
 */
public class Consumer extends Thread {

    private Container container;

    public Consumer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.get();
        }
    }
}
/**
 * 測試類
 */
public class MyThreadTest {

    public static void main(String[] args) {
        Container container = new Container();

        Producer producer = new Producer(container);
        Consumer consumer = new Consumer(container);

        producer.start();
        consumer.start();
    }
}

運行結(jié)果如下：

生產(chǎn)者：Thread-0，add：0
生產(chǎn)者：Thread-0，add：1
生產(chǎn)者：Thread-0，add：2
生產(chǎn)者：Thread-0，add：3
生產(chǎn)者：Thread-0，add：4
生產(chǎn)者：Thread-0，add：5
消費者：Thread-1，value：0
消費者：Thread-1，value：1
消費者：Thread-1，value：2
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5

可以很清晰的看到，生產(chǎn)者線程執(zhí)行完畢之后，消費者線程才開始消費。

3.2、LinkedBlockingQueue

LinkedBlockingQueue是一個基于鏈表的阻塞隊列，初始化的時候無須指定隊列大小，默認(rèn)隊列長度為Integer.MAX_VALUE，也就是 int 型最大值。

同樣的，采用的是ReentrantLock和Condition實現(xiàn)生產(chǎn)者和消費者模型，不同的是它使用了兩個lock，這意味著生產(chǎn)者和消費者可以并行運行，程序執(zhí)行效率進(jìn)一步得到提升。

部分核心源碼如下：

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    /** 使用出隊獨占鎖ReetrantLock */
    private final ReentrantLock takeLock = new ReentrantLock();

    /** 等待出隊的條件 */
    private final Condition notEmpty = takeLock.newCondition();

    /** 使用入隊獨占鎖ReetrantLock */
    private final ReentrantLock putLock = new ReentrantLock();

    /** 等待入隊的條件 */
    private final Condition notFull = putLock.newCondition();

    /**入隊操作*/
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            while (count.get() == capacity) {
                notFull.await();
            }
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

    /**出隊操作*/
    public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
            while (count.get() == 0) {
                notEmpty.await();
            }
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }
}

把最上面的樣例Container中的阻塞隊列實現(xiàn)類換成LinkedBlockingQueue，調(diào)整如下：

/**
 * 初始化阻塞隊列
 */
private final BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();

再次運行結(jié)果如下：

生產(chǎn)者：Thread-0，add：0
消費者：Thread-1，value：0
生產(chǎn)者：Thread-0，add：1
消費者：Thread-1，value：1
生產(chǎn)者：Thread-0，add：2
消費者：Thread-1，value：2
生產(chǎn)者：Thread-0，add：3
生產(chǎn)者：Thread-0，add：4
生產(chǎn)者：Thread-0，add：5
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5

可以很清晰的看到，生產(chǎn)者線程和消費者線程，交替并行執(zhí)行。

3.3、SynchronousQueue

SynchronousQueue是一個沒有緩沖的隊列，生產(chǎn)者產(chǎn)生的數(shù)據(jù)直接會被消費者獲取并且立刻消費，相當(dāng)于傳統(tǒng)的一個請求對應(yīng)一個應(yīng)答模式。

相比ArrayBlockingQueue和LinkedBlockingQueue，SynchronousQueue實現(xiàn)機制也不同，它主要采用隊列和棧來實現(xiàn)數(shù)據(jù)的傳遞，中間不存儲任何數(shù)據(jù)，生產(chǎn)的數(shù)據(jù)必須得消費者處理，線程阻塞方式采用 JDK 提供的LockSupport park/unpark函數(shù)來完成，也支持公平和非公平兩種模式。

  • 當(dāng)采用公平模式時：使用一個 FIFO 隊列來管理多余的生產(chǎn)者和消費者
  • 當(dāng)采用非公平模式時：使用一個 LIFO 棧來管理多余的生產(chǎn)者和消費者，這也是SynchronousQueue默認(rèn)的模式

部分核心源碼如下：

public class SynchronousQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {

    /**不同的策略實現(xiàn)*/
    private transient volatile Transferer<E> transferer;

 /**默認(rèn)非公平模式*/
    public SynchronousQueue() {
        this(false);
    }

    /**可以選策略，也可以采用公平模式*/
    public SynchronousQueue(boolean fair) {
        transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
    }

 /**入隊操作*/
    public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        if (transferer.transfer(e, false, 0) == null) {
            Thread.interrupted();
            throw new InterruptedException();
        }
    }

    /**出隊操作*/
    public E take() throws InterruptedException {
        E e = transferer.transfer(null, false, 0);
        if (e != null)
            return e;
        Thread.interrupted();
        throw new InterruptedException();
    }
}

同樣的，把最上面的樣例Container中的阻塞隊列實現(xiàn)類換成SynchronousQueue，代碼如下：

public class Container {

    /**
     * 初始化阻塞隊列
     */
    private final BlockingQueue<Integer> queue = new SynchronousQueue<>();


    /**
     * 添加數(shù)據(jù)到阻塞隊列
     * @param value
     */
    public void add(Integer value) {
        try {
            queue.put(value);
            Thread.sleep(100);
            System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 從阻塞隊列獲取數(shù)據(jù)
     */
    public void get() {
        try {
            Integer value = queue.take();
            Thread.sleep(200);
            System.out.println("消費者："+ Thread.currentThread().getName()+"，value：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

再次運行結(jié)果如下：

生產(chǎn)者：Thread-0，add：0
消費者：Thread-1，value：0
生產(chǎn)者：Thread-0，add：1
消費者：Thread-1，value：1
生產(chǎn)者：Thread-0，add：2
消費者：Thread-1，value：2
生產(chǎn)者：Thread-0，add：3
消費者：Thread-1，value：3
生產(chǎn)者：Thread-0，add：4
消費者：Thread-1，value：4
生產(chǎn)者：Thread-0，add：5
消費者：Thread-1，value：5

可以很清晰的看到，生產(chǎn)者線程和消費者線程，交替串行執(zhí)行，生產(chǎn)者每投遞一條數(shù)據(jù)，消費者處理一條數(shù)據(jù)。

3.4、PriorityBlockingQueue

PriorityBlockingQueue是一個基于優(yōu)先級別的阻塞隊列，底層基于數(shù)組實現(xiàn)，可以認(rèn)為是一個無界隊列。

PriorityBlockingQueue與ArrayBlockingQueue的實現(xiàn)邏輯，基本相似，也是采用ReentrantLock來實現(xiàn)加鎖的操作。

最大不同點在于：

  • PriorityBlockingQueue內(nèi)部基于數(shù)組實現(xiàn)的最小二叉堆算法，可以對隊列中的元素進(jìn)行排序，插入隊列的元素需要實現(xiàn)Comparator或者Comparable接口，以便對元素進(jìn)行排序
  • 其次，隊列的長度是可擴展的，不需要顯式指定長度，上限為Integer.MAX_VALUE - 8

部分核心源碼如下：

public class PriorityBlockingQueue<E> extends AbstractQueue<E>
    implements BlockingQueue<E>, java.io.Serializable {

  /**隊列元素*/
    private transient Object[] queue;

    /**比較器*/
    private transient Comparator<? super E> comparator;

    /**采用ReentrantLock進(jìn)行加鎖*/
    private final ReentrantLock lock;

    /**條件等待與通知*/
    private final Condition notEmpty;

    /**入隊操作*/
    public boolean offer(E e) {
        if (e == null)
            throw new NullPointerException();
        final ReentrantLock lock = this.lock;
        lock.lock();
        int n, cap;
        Object[] array;
        while ((n = size) >= (cap = (array = queue).length))
            tryGrow(array, cap);
        try {
            Comparator<? super E> cmp = comparator;
            if (cmp == null)
                siftUpComparable(n, e, array);
            else
                siftUpUsingComparator(n, e, array, cmp);
            size = n + 1;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
        return true;
    }

    /**出隊操作*/
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        E result;
        try {
            while ( (result = dequeue()) == null)
                notEmpty.await();
        } finally {
            lock.unlock();
        }
        return result;
    }
}

同樣的，把最上面的樣例Container中的阻塞隊列實現(xiàn)類換成PriorityBlockingQueue，調(diào)整如下：

/**
 * 初始化阻塞隊列
 */
private final BlockingQueue<Integer> queue = new PriorityBlockingQueue<>();

生產(chǎn)者插入數(shù)據(jù)的內(nèi)容，我們改下插入順序。

/**
 * 生產(chǎn)者
 */
public class Producer extends Thread {

    private Container container;

    public Producer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        container.add(5);
        container.add(3);
        container.add(1);
        container.add(2);
        container.add(0);
        container.add(4);
    }
}

最后運行結(jié)果如下：

生產(chǎn)者：Thread-0，add：5
生產(chǎn)者：Thread-0，add：3
生產(chǎn)者：Thread-0，add：1
生產(chǎn)者：Thread-0，add：2
生產(chǎn)者：Thread-0，add：0
生產(chǎn)者：Thread-0，add：4
消費者：Thread-1，value：0
消費者：Thread-1，value：1
消費者：Thread-1，value：2
消費者：Thread-1，value：3
消費者：Thread-1，value：4
消費者：Thread-1，value：5

從日志上可以很明顯看出，對于整數(shù)，默認(rèn)情況下，按照升序排序，消費者默認(rèn)從 0 開始處理。

3.5、DelayQueue

DelayQueue是一個線程安全的延遲隊列，存入隊列的元素不會立刻被消費，只有到了其指定的延遲時間，才能夠從隊列中出隊。

底層采用的是PriorityQueue來存儲元素，DelayQueue的特點在于：插入隊列中的數(shù)據(jù)可以按照自定義的delay時間進(jìn)行排序，快到期的元素會排列在前面，只有delay時間小于 0 的元素才能夠被取出。

部分核心源碼如下：

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {

    /**采用ReentrantLock進(jìn)行加鎖*/
    private final transient ReentrantLock lock = new ReentrantLock();

    /**采用PriorityQueue進(jìn)行存儲數(shù)據(jù)*/
    private final PriorityQueue<E> q = new PriorityQueue<E>();

 /**條件等待與通知*/
    private final Condition available = lock.newCondition();

    /**入隊操作*/
    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

    /**出隊操作*/
    public E poll() {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            E first = q.peek();
            if (first == null || first.getDelay(NANOSECONDS) > 0)
                return null;
            else
                return q.poll();
        } finally {
            lock.unlock();
        }
    }
}

同樣的，把最上面的樣例Container中的阻塞隊列實現(xiàn)類換成DelayQueue，代碼如下：

public class Container {

    /**
     * 初始化阻塞隊列
     */
    private final BlockingQueue<DelayedUser> queue = new DelayQueue<DelayedUser>();


    /**
     * 添加數(shù)據(jù)到阻塞隊列
     * @param value
     */
    public void add(DelayedUser value) {
        try {
            queue.put(value);
            System.out.println("生產(chǎn)者："+ Thread.currentThread().getName()+"，add：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 從阻塞隊列獲取數(shù)據(jù)
     */
    public void get() {
        try {
            DelayedUser value = queue.take();
            String time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
            System.out.println(time + " 消費者："+ Thread.currentThread().getName()+"，value：" + value);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

DelayQueue隊列中的元素需要顯式實現(xiàn)Delayed接口，定義一個DelayedUser類，代碼如下：

public class DelayedUser implements Delayed {

    /**
     * 當(dāng)前時間戳
     */
    private long start;

    /**
     * 延遲時間(單位：毫秒)
     */
    private long delayedTime;

    /**
     * 名稱
     */
    private String name;

    public DelayedUser(long delayedTime, String name) {
        this.start = System.currentTimeMillis();
        this.delayedTime = delayedTime;
        this.name = name;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        // 獲取當(dāng)前延遲的時間
        long diffTime = (start + delayedTime) - System.currentTimeMillis();
        return unit.convert(diffTime,TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        // 判斷當(dāng)前對象的延遲時間是否大于目標(biāo)對象的延遲時間
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }

    @Override
    public String toString() {
        return "DelayedUser{" +
                "delayedTime=" + delayedTime +
                ", name='" + name + '\'' +
                '}';
    }
}

生產(chǎn)者插入數(shù)據(jù)的內(nèi)容，做如下調(diào)整。

/**
 * 生產(chǎn)者
 */
public class Producer extends Thread {

    private Container container;

    public Producer(Container container) {
        this.container = container;
    }

    @Override
    public void run() {
        for (int i = 0; i < 6; i++) {
            container.add(new DelayedUser(1000 * i, "張三" +  i));
        }
    }
}

最后運行結(jié)果如下：

生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=0, name='張三0'}
生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=1000, name='張三1'}
生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=2000, name='張三2'}
生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=3000, name='張三3'}
生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=4000, name='張三4'}
生產(chǎn)者：Thread-0，add：DelayedUser{delayedTime=5000, name='張三5'}
2023-11-03 14:55:33 消費者：Thread-1，value：DelayedUser{delayedTime=0, name='張三0'}
2023-11-03 14:55:34 消費者：Thread-1，value：DelayedUser{delayedTime=1000, name='張三1'}
2023-11-03 14:55:35 消費者：Thread-1，value：DelayedUser{delayedTime=2000, name='張三2'}
2023-11-03 14:55:36 消費者：Thread-1，value：DelayedUser{delayedTime=3000, name='張三3'}
2023-11-03 14:55:37 消費者：Thread-1，value：DelayedUser{delayedTime=4000, name='張三4'}
2023-11-03 14:55:38 消費者：Thread-1，value：DelayedUser{delayedTime=5000, name='張三5'}

可以很清晰的看到，延遲時間最低的排在最前面。

04、小結(jié)

最后我們來總結(jié)一下BlockingQueue阻塞隊列接口，它提供了很多非常豐富的生產(chǎn)者和消費者模型的編程實現(xiàn)，同時兼顧了線程安全和執(zhí)行效率的特點。

開發(fā)者可以通過BlockingQueue阻塞隊列接口，簡單的代碼編程即可實現(xiàn)多線程中數(shù)據(jù)高效安全傳輸?shù)哪康模_切的說，它幫助開發(fā)者減輕了不少的編程難度。

在實際的業(yè)務(wù)開發(fā)中，其中LinkedBlockingQueue使用的是最廣泛的，因為它的執(zhí)行效率最高，在使用的時候，需要平衡好隊列長度，防止過大導(dǎo)致內(nèi)存溢出。

舉個最簡單的例子，比如某個功能上線之后，需要做下壓力測試，總共需要請求 10000 次，采用 100 個線程去執(zhí)行，測試服務(wù)是否能正常工作。如何實現(xiàn)呢？

可能有的同學(xué)想到，每個線程執(zhí)行 100 次請求，啟動 100 個線程去執(zhí)行，可以是可以，就是有點笨拙。

其實還有另一個辦法，就是將 10000 個請求對象，存入到阻塞隊列中，然后采用 100 個線程去消費執(zhí)行，這種編程模型會更佳靈活。

具體示例代碼如下：

public static void main(String[] args) throws InterruptedException {
    // 將每個用戶訪問百度服務(wù)的請求任務(wù)，存入阻塞隊列中
    // 也可以也采用多線程寫入
    BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    for (int i = 0; i < 10000; i++) {
        queue.put("https://www.baidu.com?paramKey=" + i);
    }

    // 模擬100個線程，執(zhí)行10000次請求訪問百度
    final int threadNum = 100;
    for (int i = 0; i < threadNum; i++) {
        final int threadCount = i + 1;
        new Thread(new Runnable() {

            @Override
            public void run() {
                System.out.println("thread " + threadCount + " start");
                boolean over = false;
                while (!over) {
                    String url = queue.poll();
                    if(Objects.nonNull(url)) {
                        // 發(fā)起請求
                        String result =HttpUtils.getUrl(url);
                        System.out.println("thread " + threadCount + " run result:" + result);
                    }else {
                        // 任務(wù)結(jié)束
                        over = true;
                        System.out.println("thread " + threadCount + " final");
                    }
                }
            }
        }).start();
    }
}

本文主要圍繞BlockingQueue阻塞隊列接口，從方法介紹到用法詳解，做了一次知識總結(jié)，如果有描述不對的地方，歡迎留言指出！

05、參考

1、https://www.cnblogs.com/xrq730/p/4855857.html

2、https://juejin.cn/post/6999798721269465102

責(zé)任編輯：武曉燕 來源： 潘志的研發(fā)筆記
阻塞接口開發(fā)者
相關(guān)推薦

2022-09-09 08:08:28

開源項目服務(wù)

2023-07-31 08:21:22

語法校對器Pick

2020-11-30 13:10:39

MySQL安全服務(wù)器

2021-08-06 11:50:49

Linux 字節(jié)對齊Linux 系統(tǒng)

2023-07-03 09:59:00

并發(fā)編程并發(fā)容器

2019-01-07 12:02:02

TCP長連接Java

2024-02-21 07:24:21

微服務(wù)單體架構(gòu)MVC

2016-11-25 00:45:37

隊列數(shù)據(jù)

2018-05-31 13:50:30

Java高并發(fā)

2017-12-21 08:06:40

2011-06-15 16:36:27

Qt 圖形

2016-11-28 09:08:43

java系統(tǒng)異步非阻塞

2017-01-10 13:33:51

iOS編程throttle

2020-10-30 07:43:35

Jenkins配置前端

2019-01-28 18:02:21

網(wǎng)絡(luò)安全網(wǎng)絡(luò)安全技術(shù)周刊

2011-07-15 10:27:21

PHP

2021-10-26 11:42:51

系統(tǒng)

2024-05-06 12:35:26

C#編程遞歸

2022-03-09 09:43:20

并發(fā)編程Java

2022-02-15 19:50:18

加載器JAVAJDK
點贊
收藏

51CTO技術(shù)棧公眾號