手把手帶你理解java線程池之工作隊(duì)列workQueue

更新時(shí)間：2021年09月10日 09:20:02 作者：渣男小四

這篇文章主要介紹了java線程池之工作隊(duì)列workQueue，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值，需要的朋友可以參考下

線程池之工作隊(duì)列

ArrayBlockingQueue

采用數(shù)組來(lái)實(shí)現(xiàn),并采用可重入鎖ReentrantLock來(lái)做并發(fā)控制，無(wú)論是添加還是讀取，都先要獲得鎖才能進(jìn)行操作可看出進(jìn)行讀寫操作都使用了ReentrantLock，ArrayBlockingQueue需要為其指定容量

public boolean offer(E e) {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            if (count == items.length)
                return false;
            else {
                enqueue(e);
                return true;
            }
        } finally {
            lock.unlock();
        }
    }
    
    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();
        }
    }

SynchronousQueue

由于SynchronousQueue源碼比較復(fù)雜，里面大量的Cas操作，SynchronousQueue沒(méi)有容器，所以里面是裝不了任務(wù)的，當(dāng)一個(gè)生產(chǎn)者線程生產(chǎn)一個(gè)任務(wù)的時(shí)候，如果沒(méi)有對(duì)應(yīng)的消費(fèi)者消費(fèi)，那么該生產(chǎn)者會(huì)一直阻塞，知道有消費(fèi)者消費(fèi)為止。
圖示：

如下代碼，如果我們將消費(fèi)者線程注釋掉執(zhí)行，那么生產(chǎn)者哪里將會(huì)一直阻塞

package thread.customthreadpool;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 測(cè)試SynchronousQueue
 */
public class SynchronousQueueTest {

    private static final SynchronousQueue<String> synchronousQueue = new SynchronousQueue<>();

    private static final ExecutorService service = Executors.newCachedThreadPool();

    public static void main(String[] args) {
        /**
         * Provider
         */
        service.submit(() -> {
            try {
                synchronousQueue.put("liu");
            }catch (Exception e){
                e.printStackTrace();
            }
            System.out.println("Consumer finished spending");
        });

        /**
         * Consumer
         */
        service.submit(() ->{
            try {
                synchronousQueue.take();
            }catch (Exception e){
                e.printStackTrace();
            }
            System.out.println("take over");
        });
    }
}

LinkedBlockingDeque

LinkedBlockingDeque是一個(gè)雙向隊(duì)列，底層使用單鏈表實(shí)現(xiàn)，任何一段都可進(jìn)行元素的讀寫操作，在初始化LinkedBlockingDeque的時(shí)候，我們可以指定容量，也可不指定，如果不指定，則容量為Integer.MAX_VALUE，

注：Deque是雙端隊(duì)列，而Queue是單端隊(duì)列，雙端意思是兩端都可以進(jìn)行讀寫操作，而單端則只能從一端進(jìn)，一端出（FIFO）

public LinkedBlockingDeque() {
        this(Integer.MAX_VALUE);
}

package thread.customthreadpool;
import java.util.concurrent.LinkedBlockingDeque;
public class LinkedBlockingDequeTest {

    private static final LinkedBlockingDeque<Integer> deque = new LinkedBlockingDeque<>();

    public static void main(String[] args) throws InterruptedException {
        deque.put(1);
        deque.put(2);
        deque.put(3);
        deque.put(4);
        deque.put(5);
        System.out.println(deque);
        System.out.println("deque size  "+deque.size());
        deque.take();
        deque.take();
        deque.take();
        deque.take();
        deque.take();
        System.out.println(deque);
        System.out.println("deque size  "+deque.size());
    }
}

LinkedBlockingQueue

底層基于單向連表實(shí)現(xiàn)，是一個(gè)單向隊(duì)列，具有先進(jìn)先出(FIFO)特點(diǎn)，使用了ReentrantLock來(lái)做并發(fā)控制,讀寫操作都上鎖

private final ReentrantLock putLock = new ReentrantLock();
    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;
    }

DelayDeque

DelayDeque是一個(gè)無(wú)界隊(duì)列，添加進(jìn)DelayDeque的元素會(huì)經(jīng)過(guò)compareTo方法計(jì)算，然后按照時(shí)間進(jìn)行排序，排在隊(duì)頭的元素是最早到期的，越往后到期時(shí)間越長(zhǎng)，DelayDeque只能接受Delayed接口類型如圖所示，隊(duì)列里的元素并不是按照先進(jìn)先出的規(guī)則，而是按照過(guò)期時(shí)間

示例

package thread.customthreadpool.delayDeque;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class MyDelayed implements Delayed {

    private final String taskName ;
    private final long nowTime = System.currentTimeMillis();
    private final long expireTime ;

    public MyDelayed(String taskName,long expireTime) {
        this.taskName = taskName;
        this.expireTime = expireTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert((nowTime+expireTime) - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed o) {
        MyDelayed myDelayed = (MyDelayed) o;
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }

    @Override
    public String toString() {
        return "MyDelayed{" +
                "taskName='" + taskName + '\'' +
                ", nowTime=" + nowTime +
                ", expireTime=" + expireTime +
                '}';
    }
}

package thread.customthreadpool.delayDeque;

import java.util.concurrent.*;

public class MyDelayQueue {

    private static final DelayQueue<MyDelayed> delayQueue = new DelayQueue<>();

    private static final ExecutorService service = Executors.newCachedThreadPool();

    public static void main(String[] args) throws InterruptedException {
        service.submit(() -> {
            delayQueue.put(new MyDelayed("A-Task",5000));
            delayQueue.put(new MyDelayed("B-Task",4000));
            delayQueue.put(new MyDelayed("C-Task",3000));
            delayQueue.put(new MyDelayed("D-Task",2000));
            delayQueue.put(new MyDelayed("E-Task",1000));
        });
        while (true){
            System.out.println(delayQueue.take());
        }
    }
}

result

應(yīng)用場(chǎng)景

1.美團(tuán)外賣訂單：當(dāng)我們下單后沒(méi)付款，30分鐘后將自動(dòng)取消訂單
2.緩存，對(duì)于某些任務(wù)，需要在特定的時(shí)間清理；

and so on

LinkedTransferQueue

當(dāng)消費(fèi)線程從隊(duì)列中取元素時(shí)，如果隊(duì)列為空，那么生成一個(gè)為null的節(jié)點(diǎn)，消費(fèi)者線程就一直等待，此時(shí)如果生產(chǎn)者線程發(fā)現(xiàn)隊(duì)列中有一個(gè)null節(jié)點(diǎn)，它就不入隊(duì)了，而是將元素填充到這個(gè)null節(jié)點(diǎn)并喚醒消費(fèi)者線程，然后消費(fèi)者線程取走元素。
LinkedTransferQueue是 SynchronousQueue 和 LinkedBlockingQueue 的整合，性能比較高，因?yàn)闆](méi)有鎖操作， SynchronousQueue不能存儲(chǔ)元素，而LinkedTransferQueue能存儲(chǔ)元素，

PriorityBlockingQueue

PriorityBlockingQueue是一個(gè)無(wú)界的阻塞隊(duì)列，同時(shí)是一個(gè)支持優(yōu)先級(jí)的隊(duì)列，讀寫操作都是基于ReentrantLock, 內(nèi)部使用堆算法保證每次出隊(duì)都是優(yōu)先級(jí)最高的元素

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;
}

到此這篇關(guān)于手把手帶你理解java線程池之工作隊(duì)列workQueue的文章就介紹到這了,更多相關(guān)java線程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: