欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Java阻塞延遲隊列DelayQueue原理及使用詳解

 更新時間:2023年12月08日 08:58:20   作者:GeorgiaStar  
這篇文章主要介紹了Java阻塞延遲隊列DelayQueue原理及使用詳解,阻塞隊列是一個支持兩個附加操作的隊列,這兩個附加的操作是:在隊列為空時,從隊列中獲取元素的消費者線程會一直等待直到隊列變?yōu)榉强?需要的朋友可以參考下

前言

從阻塞隊列說起,阻塞隊列(BlockingQueue)是一個支持兩個附加操作的隊列。

這兩個附加的操作是:在隊列為空時,從隊列中獲取元素的消費者線程會一直等待直到隊列變?yōu)榉强铡?/p>

當(dāng)隊列滿時,向隊列中放置元素的生產(chǎn)者線程會等待直到隊列可用。

阻塞隊列常用于生產(chǎn)者和消費者的場景,生產(chǎn)者是往隊列里添加元素的線程,消費者是從隊列里拿元素的線程。

在阻塞隊列不可用時,這兩個附加操作提供了4種處理方式:

  • 拋出異常:當(dāng)隊列滿時,插入元素會拋出IllegalStateException;
  • 返回特殊值:offer()是入隊方法,當(dāng)插入成功時返回true,插入失敗返回false;poll()是出隊方法,當(dāng)出隊成功時返回元素的值,隊列為空時返回null
  • 一直阻塞:當(dāng)隊列滿時,阻塞執(zhí)行插入方法的線程;當(dāng)隊列空時,阻塞執(zhí)行出隊方法的線程
  • 超時退出:顧名思義

下面是Java常見的阻塞隊列。

  • ArrayBlockingQueue :一個由數(shù)組結(jié)構(gòu)組成的有界阻塞隊列
  • LinkedBlockingQueue :一個由鏈表結(jié)構(gòu)組成的有界阻塞隊列
  • PriorityBlockingQueue :一個支持優(yōu)先級排序的無界阻塞隊列
  • DelayQueue:一個使用優(yōu)先級隊列實現(xiàn)的無界阻塞隊列
  • SynchronousQueue:一個不存儲元素的阻塞隊列
  • LinkedTransferQueue:一個由鏈表結(jié)構(gòu)組成的無界阻塞隊列
  • LinkedBlockingDeque:一個由鏈表結(jié)構(gòu)組成的雙向阻塞隊列

DelayQueue解析

DelayQueue隊列中每個元素都有一個過期時間,并且隊列是個優(yōu)先級隊列,當(dāng)從隊列獲取元素的時候,只有過期元素才會出隊,DelayQueue的類結(jié)構(gòu)如下圖所示:

如圖DelayQueue中內(nèi)部使用的是PriorityQueue存放數(shù)據(jù),使用ReentrantLock實現(xiàn)線程同步。

另外隊列里面的元素要實現(xiàn)Delayed接口,一個是獲取當(dāng)前剩余時間的接口,一個是元素比較的接口,因為這個是有優(yōu)先級的隊列。

DelayQueue類的主要成員

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    implements BlockingQueue<E> {
    // 持有內(nèi)部重入鎖。
    private final transient ReentrantLock lock = new ReentrantLock();
    // 優(yōu)先級隊列,存放工作任務(wù)。
    private final PriorityQueue<E> q = new PriorityQueue<E>();
    private Thread leader = null;
    // 依賴于重入鎖的condition。
    private final Condition available = lock.newCondition();
}

元素入隊列

插入元素到隊列中主要三個方法,但實際上底層調(diào)用的都是offer(e)方法

/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true} (as specified by {@link Collection#add})
 * @throws NullPointerException if the specified element is null
 */
public boolean add(E e) {
    return offer(e);
}
/**
 * Inserts the specified element into this delay queue. As the queue is
 * unbounded this method will never block.
 *
 * @param e the element to add
 * @throws NullPointerException {@inheritDoc}
 */
public void put(E e) {
    offer(e);
}
/**
 * Inserts the specified element into this delay queue.
 *
 * @param e the element to add
 * @return {@code true}
 * @throws NullPointerException if the specified element is null
 */
public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    //獲取到重入鎖
    lock.lock();
    try {
        q.offer(e);
        //添加成功元素
        if (q.peek() == e) {
            leader = null;
            //將等待隊列中的頭節(jié)點移動到同步隊列。
            available.signal();
        }
        return true;
    } finally {
        lock.unlock();
    }
}

首先獲取獨占鎖,然后添加元素到優(yōu)先級隊列,由于q是優(yōu)先級隊列,所以添加完元素后,peek()方法返回的并不一定是剛才添加的元素,如果判斷為true,說明當(dāng)前元素e的優(yōu)先級最小也就是即將過期的,這時候激活avaliable變量條件隊列里面的線程,通知它們隊列里面有元素了。

從隊列中取元素

有兩個方法可以取元素(都是取隊頭),poll()方法取隊頭當(dāng)隊頭元素沒過期時返回null,take()方法取隊頭當(dāng)隊頭元素沒過期時會一直等待。

/**
 * Retrieves and removes the head of this queue, or returns {@code null}
 * if this queue has no elements with an expired delay.
 *
 * @return the head of this queue, or {@code null} if this
 *         queue has no elements with an expired delay
 */
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        E first = q.peek();
        //如果隊列為空,或者不為空但是隊頭元素沒有過期則返回null
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            return null;
        else
            return q.poll();
    } finally {
        lock.unlock();
    }
}
/**
 * Retrieves and removes the head of this queue, waiting if necessary
 * until an element with an expired delay is available on this queue.
 *
 * @return the head of this queue
 * @throws InterruptedException {@inheritDoc}
 */
public E take() throws InterruptedException {
    // 獲取鎖。每個延遲隊列內(nèi)聚了一個重入鎖。
    final ReentrantLock lock = this.lock;
    // 獲取可中斷的鎖。
    lock.lockInterruptibly();
    try {
        for (;;) {
            // 嘗試從優(yōu)先級隊列中獲取隊列頭部元素,獲取但不移除
            E first = q.peek();
            if (first == null)
                //無元素,當(dāng)前線程節(jié)點加入等待隊列,并阻塞當(dāng)前線程
                available.await();
            else {
                // 通過延遲任務(wù)的getDelay()方法獲取延遲時間
                long delay = first.getDelay(NANOSECONDS);
                if (delay <= 0)
                    //延遲時間到期,獲取并刪除頭部元素。
                    return q.poll();
                first = null; // don't retain ref while waiting
                if (leader != null)
                    available.await();
                else {
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 線程節(jié)點進入等待隊列 x 納秒。
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 若還存在元素的話,則將等待隊列頭節(jié)點中的線程節(jié)點移動到同步隊列中。
        if (leader == null && q.peek() != null)
            available.signal();
        lock.unlock();
    }
}

重點說一下take()方法,第一次調(diào)用take時候由于隊列空,所以把當(dāng)前線程放入available的條件隊列中等待,當(dāng)執(zhí)行offer()成功并且添加的新元素恰好就是優(yōu)先級隊列的隊首時就會通知最先等待的線程激活,循環(huán)重新獲取隊首元素,這時候first假如不空,則調(diào)用getDelay()方法看該元素還剩下多少時間就過期了,如果delay<=0則說明已經(jīng)過期,則直接出隊返回。

否則看leader是否為null,不為null則說明是其他線程也在執(zhí)行take()則把當(dāng)前線程放入條件隊列,否則就是只有當(dāng)前線程在執(zhí)行take()方法,則當(dāng)前線程await直到剩余過期時間到,這期間該線程會釋放鎖,所以其他線程可以offer()添加元素,也可以take()阻塞自己,剩余過期時間到后,當(dāng)前線程會重新競爭鎖,重新進入循環(huán)。

如果已經(jīng)具備了JUC包中的Lock接口以及AQS的相關(guān)知識,上述代碼大部分應(yīng)該都比較容易理解。

DelayQueue將實現(xiàn)了Delayed接口的對象添加到優(yōu)先級隊列中,通過在可重入鎖的Condition上調(diào)用await()方法,實現(xiàn)了延遲獲取阻塞隊列中元素的功能。

總結(jié)

  • DelayQueue是一個內(nèi)部依靠AQS隊列同步器所實現(xiàn)的無界延遲阻塞隊列。
  • 隊列中的延遲對象需要覆蓋getDelay()與compareTo()方法,并且要注意 getDelay()的時間單位的統(tǒng)一,compareTo()根據(jù)業(yè)務(wù)邏輯進行合理的比較邏輯重寫。
  • DelayQueue中內(nèi)聚的可重入鎖是非公平的。
  • 延遲隊列是實現(xiàn)定時任務(wù)的關(guān)鍵,ScheduledThreadPoolExecutor中的任務(wù)隊列是DelayedWorkQueue,其和DelayedQueue高度類似,也是一個延遲隊列。

DelayQueue使用例子

寫一個簡單的例子:

public class DelayQueueTest {
    public static final int SIZE = 10;
    public static void main(String[] args) {
    	DelayQueueTest test = new DelayQueueTest();
        //初始化線程池
        BlockingQueue<Runnable> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
        ThreadPoolExecutor threadPool = new ThreadPoolExecutor
            (5, 10, 10, TimeUnit.MILLISECONDS,
                arrayBlockingQueue, Executors.defaultThreadFactory(),
                new ThreadPoolExecutor.AbortPolicy());
        DelayQueue<DelayedTask> delayTaskQueue = new DelayQueue<>();
        //模擬SIZE個延遲任務(wù)
        for (byte i = 0; i < SIZE; i++) {
            Long runAt = System.currentTimeMillis() + 1000 * i;
            String name = "Zhang_" + i;
            byte age = (byte)(10 + i);
            String gender = (i % 2 == 0 ? "male" : "female");
            Student student = new StudentBuilder(name, age, gender).height(150 + i).province("ZheJiang").build();
            delayTaskQueue.put(new DelayedTask<Student>(student, 1, function -> test.print(student), runAt));
        }
        while (true) {
            if (delayTaskQueue.size() == 0) {
                break;
            }
            try {
                //從延遲隊列中取值,如果沒有對象過期則取到null
                DelayedTask delayedTask = delayTaskQueue.poll();
                if (delayedTask != null) {
                    threadPool.execute(delayedTask);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        threadPool.shutdown();
    }
    public String print(Object object) {
      System.out.println(Thread.currentThread().getName());
        String str = ">>>junit log>>>" + object.getClass().getSimpleName() + ":" + object.toString();
        System.out.println(str);
        return str;
    }
    private static class DelayedTask<T> implements Delayed, Runnable {
        /**
         * 任務(wù)參數(shù)
         */
        private T taskParam;
        /**
         * 任務(wù)類型
         */
        private Integer type;
        /**
         * 任務(wù)函數(shù)
         */
        private Function<T, String> function;
        /**
         * 任務(wù)執(zhí)行時刻
         */
        private Long runAt;
        public T getTaskParam() {
            return taskParam;
        }
        public Integer getType() {
            return type;
        }
        public Function<T, String> getFunction() {
            return function;
        }
        public Long getRunAt() {
            return runAt;
        }
        DelayedTask(T taskParam, Integer type, Function<T, String> function, Long runAt) {
            this.taskParam = taskParam;
            this.type = type;
            this.function = function;
            this.runAt = runAt;
        }
        @Override
        public void run() {
            if (taskParam != null) {
                function.apply(taskParam);
            }
        }
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.runAt - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        @Override
        public int compareTo(Delayed o) {
            DelayedTask object = (DelayedTask)o;
            return this.runAt.compareTo(object.getRunAt());
        }
    }
}

運行結(jié)果如下,由于10個元素的延遲時間均相差1秒,可以看到逐步打印的效果。

DelayQueue典型場景是重試機制實現(xiàn),比如當(dāng)調(diào)用接口失敗后,把當(dāng)前調(diào)用信息放入delay=10s的元素,然后把元素放入隊列,那么這個隊列就是一個重試隊列,一個線程通過take()方法獲取需要重試的接口,take()返回則接口進行重試,失敗則再次放入隊列,同時也可以在元素加上重試次數(shù)。

到此這篇關(guān)于Java阻塞延遲隊列DelayQueue原理及使用詳解的文章就介紹到這了,更多相關(guān)阻塞延遲隊列DelayQueue原理內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

  • Maven中optional標(biāo)簽用法詳解

    Maven中optional標(biāo)簽用法詳解

    這篇文章主要介紹了Maven中optional標(biāo)簽,文章中有詳細的代碼示例供大家參考,對大家的學(xué)習(xí)或工作有一定的參考價值,感興趣的小伙伴可以借鑒一下
    2023-05-05
  • SpringBoot如何讀取resources目錄下的文件

    SpringBoot如何讀取resources目錄下的文件

    這篇文章主要介紹了SpringBoot如何讀取resources目錄下的文件問題,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2023-07-07
  • kafka并發(fā)寫大消息異常TimeoutException排查記錄

    kafka并發(fā)寫大消息異常TimeoutException排查記錄

    這篇文章主要為大家介紹了kafka并發(fā)寫大消息異常TimeoutException的排查記錄及解決方案,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步
    2022-02-02
  • Java并發(fā)編程之Fork/Join框架詳解

    Java并發(fā)編程之Fork/Join框架詳解

    這篇文章主要介紹了Java并發(fā)編程之Fork/Join框架詳解,Fork/Join框架是Java7提供的一個用于并行執(zhí)行任務(wù)的框架,是一個把大任務(wù)分割成若干個小任務(wù),最終匯總每個小任務(wù)結(jié)果后得到大任務(wù)結(jié)果的框架,需要的朋友可以參考下
    2023-12-12
  • 詳解SpringBoot實現(xiàn)fastdfs防盜鏈功能的示例代碼

    詳解SpringBoot實現(xiàn)fastdfs防盜鏈功能的示例代碼

    我們可以通過fastdfs實現(xiàn)一個分布式文件系統(tǒng),如果fastdfs部署在外網(wǎng),那么任何一個人知道了上傳接口,就可以實現(xiàn)文件的上傳和訪問。那么如何阻止他人訪問我們fastdfs服務(wù)器上的文件呢?此處就需要使用fastdfs的防盜鏈功能,本文就來講講如何實現(xiàn)這一功能
    2022-10-10
  • Java8并行流中自定義線程池操作示例

    Java8并行流中自定義線程池操作示例

    這篇文章主要介紹了Java8并行流中自定義線程池操作,結(jié)合實例形式分析了并行流的相關(guān)概念、定義及自定義線程池的相關(guān)操作技巧,需要的朋友可以參考下
    2019-05-05
  • Java中包裝類介紹與其注意事項

    Java中包裝類介紹與其注意事項

    Java語言是一個面向?qū)ο蟮恼Z言,但是Java中的基本數(shù)據(jù)類型卻是不面向?qū)ο蟮模@在實際使用時存在很多的不便,所以在設(shè)計類時為每個基本數(shù)據(jù)類型設(shè)計了一個對應(yīng)的類進行代表,這樣八個和基本數(shù)據(jù)類型對應(yīng)的類統(tǒng)稱為包裝類,有些地方也翻譯為外覆類或數(shù)據(jù)類型類。
    2017-02-02
  • springmvc學(xué)習(xí)筆記-返回json的日期格式問題的解決方法

    springmvc學(xué)習(xí)筆記-返回json的日期格式問題的解決方法

    本篇文章主要介紹了springmvc學(xué)習(xí)筆記-返回json的日期格式問題的解決方法,解決了日期格式的輸出,有興趣的可以了解一下。
    2017-01-01
  • 如何使用pipeline和jacoco獲取自動化測試代碼覆蓋率

    如何使用pipeline和jacoco獲取自動化測試代碼覆蓋率

    這篇文章主要介紹了如何使用pipeline和jacoco獲取自動化測試代碼覆蓋率,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下
    2020-11-11
  • Java實現(xiàn)word轉(zhuǎn)pdf并在關(guān)鍵字位置插入圖片

    Java實現(xiàn)word轉(zhuǎn)pdf并在關(guān)鍵字位置插入圖片

    這篇文章主要為大家詳細介紹了如何利用Java實現(xiàn)word轉(zhuǎn)pdf,并在word中關(guān)鍵字位置插入圖片,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下
    2024-11-11

最新評論