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

java8異步調(diào)用如何使用才是最好的方式

 更新時(shí)間:2022年01月27日 12:53:46   作者:我犟不過(guò)你  
異步調(diào)用主要用于當(dāng)前程序的執(zhí)行不用等待調(diào)用方法執(zhí)行結(jié)束就可以繼續(xù)執(zhí)行,下面這篇文章主要給大家介紹了關(guān)于java8異步調(diào)用如何使用才是最好的方式,文中通過(guò)實(shí)例代碼介紹的非常詳細(xì),需要的朋友可以參考下

一、異步調(diào)用方式分析

今天在寫(xiě)代碼的時(shí)候,想要調(diào)用異步的操作,這里我是用的java8的流式異步調(diào)用,但是使用過(guò)程中呢,發(fā)現(xiàn)這個(gè)異步方式有兩個(gè)方法,如下所示:

區(qū)別是一個(gè) 需要指定線程池,一個(gè)不需要。

  • 那么指定線程池有哪些好處呢?直觀的說(shuō)有以下兩點(diǎn)好處:

    • 可以根據(jù)我們的服務(wù)器性能,通過(guò)池的管理更好的規(guī)劃我們的線程數(shù)。
    • 可以對(duì)我們使用的線程自定義名稱,這里也是阿里java開(kāi)發(fā)規(guī)范所提到的。

1.1 java8異步調(diào)用默認(rèn)線程池方式

當(dāng)然常規(guī)使用默認(rèn)的也沒(méi)什么問(wèn)題。我們通過(guò)源碼分析下使用默認(rèn)線程池的過(guò)程。

   public static CompletableFuture<Void> runAsync(Runnable runnable) {
        return asyncRunStage(asyncPool, runnable);
    }

看下這個(gè)asyncPool是什么?

如下所示,useCommonPool如果為真,就使用ForkJoinPool.commonPool(),否則創(chuàng)建一個(gè)new ThreadPerTaskExecutor()

    private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

看看useCommonPool 是什么?

    private static final boolean useCommonPool =
        (ForkJoinPool.getCommonPoolParallelism() > 1);
  /**
    * 公共池的目標(biāo)并行度級(jí)別
    */
    public static int getCommonPoolParallelism() {
        return commonParallelism;
    }

最終這個(gè)并行級(jí)別并沒(méi)有給出默認(rèn)值

static final int commonParallelism;

通過(guò)找到這個(gè)常量的調(diào)用,我們看看是如何進(jìn)行初始化的,在ForkJoinPool中有一個(gè)靜態(tài)代碼塊,啟動(dòng)時(shí)會(huì)對(duì)commonParallelism進(jìn)行初始化,我們只關(guān)注最后一句話就好了,:

    // Unsafe mechanics
    private static final sun.misc.Unsafe U;
    private static final int  ABASE;
    private static final int  ASHIFT;
    private static final long CTL;
    private static final long RUNSTATE;
    private static final long STEALCOUNTER;
    private static final long PARKBLOCKER;
    private static final long QTOP;
    private static final long QLOCK;
    private static final long QSCANSTATE;
    private static final long QPARKER;
    private static final long QCURRENTSTEAL;
    private static final long QCURRENTJOIN;

    static {
        // initialize field offsets for CAS etc
        try {
            U = sun.misc.Unsafe.getUnsafe();
            Class<?> k = ForkJoinPool.class;
            CTL = U.objectFieldOffset
                (k.getDeclaredField("ctl"));
            RUNSTATE = U.objectFieldOffset
                (k.getDeclaredField("runState"));
            STEALCOUNTER = U.objectFieldOffset
                (k.getDeclaredField("stealCounter"));
            Class<?> tk = Thread.class;
            PARKBLOCKER = U.objectFieldOffset
                (tk.getDeclaredField("parkBlocker"));
            Class<?> wk = WorkQueue.class;
            QTOP = U.objectFieldOffset
                (wk.getDeclaredField("top"));
            QLOCK = U.objectFieldOffset
                (wk.getDeclaredField("qlock"));
            QSCANSTATE = U.objectFieldOffset
                (wk.getDeclaredField("scanState"));
            QPARKER = U.objectFieldOffset
                (wk.getDeclaredField("parker"));
            QCURRENTSTEAL = U.objectFieldOffset
                (wk.getDeclaredField("currentSteal"));
            QCURRENTJOIN = U.objectFieldOffset
                (wk.getDeclaredField("currentJoin"));
            Class<?> ak = ForkJoinTask[].class;
            ABASE = U.arrayBaseOffset(ak);
            int scale = U.arrayIndexScale(ak);
            if ((scale & (scale - 1)) != 0)
                throw new Error("data type scale not a power of two");
            ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);
        } catch (Exception e) {
            throw new Error(e);
        }

        commonMaxSpares = DEFAULT_COMMON_MAX_SPARES;
        defaultForkJoinWorkerThreadFactory =
            new DefaultForkJoinWorkerThreadFactory();
        modifyThreadPermission = new RuntimePermission("modifyThread");

        common = java.security.AccessController.doPrivileged
            (new java.security.PrivilegedAction<ForkJoinPool>() {
                public ForkJoinPool run() { return makeCommonPool(); }});
         // 即使線程被禁用也是1,至少是個(gè)1
        int par = common.config & SMASK;
        commonParallelism = par > 0 ? par : 1;
    }

如下所示,默認(rèn)是7:

所以接著下面的代碼看:

    private static final boolean useCommonPool =
        (ForkJoinPool.getCommonPoolParallelism() > 1);

這里一定是返回true,證明當(dāng)前是并行的。

    private static final Executor asyncPool = useCommonPool ?
        ForkJoinPool.commonPool() : new ThreadPerTaskExecutor();

上面會(huì)返回一個(gè)大小是七的的默認(rèn)線程池

其實(shí)這個(gè)默認(rèn)值是當(dāng)前cpu的核心數(shù),我的電腦是八核,在代碼中默認(rèn)會(huì)將核心數(shù)減一,所以顯示是七個(gè)線程。

        if (parallelism < 0 && //默認(rèn)是1,小于核心數(shù)
            (parallelism = Runtime.getRuntime().availableProcessors() - 1) <= 0)
            parallelism = 1;
        if (parallelism > MAX_CAP)
            parallelism = MAX_CAP;

下面我們寫(xiě)個(gè)main方法測(cè)試一下,10個(gè)線程,每個(gè)阻塞10秒,看結(jié)果:

    public static void main(String[] args) {
        // 創(chuàng)建10個(gè)任務(wù),每個(gè)任務(wù)阻塞10秒
        for (int i = 0; i < 10; i++) {
            CompletableFuture.runAsync(() -> {
                try {
                    Thread.sleep(10000);
                    System.out.println(new Date() + ":" + Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

結(jié)果如下所示,前面七個(gè)任務(wù)先完成,另外三個(gè)任務(wù)被阻塞10秒后,才完成:

Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-5
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-4
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-2
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-7
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-3
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-6
Mon Sep 13 11:20:57 CST 2021:ForkJoinPool.commonPool-worker-1
-----------------------------------------------------------  
Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-2
Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-5
Mon Sep 13 11:21:07 CST 2021:ForkJoinPool.commonPool-worker-4

結(jié)論:當(dāng)我們使用默認(rèn)的線程池進(jìn)行異步調(diào)用時(shí),如果異步任務(wù)是一個(gè)IO密集型,簡(jiǎn)單說(shuō)處理時(shí)間占用長(zhǎng),將導(dǎo)致其他使用共享線程池的任務(wù)阻塞,造成系統(tǒng)性能下降甚至異常。甚至當(dāng)一部分調(diào)用接口時(shí),如果接口超時(shí),那么也會(huì)阻塞與超時(shí)時(shí)長(zhǎng)相同的時(shí)間;實(shí)際在計(jì)算密集的場(chǎng)景下使用是能提高性能的。

二、使用自定義的線程池

上面說(shuō)到如果是IO密集型的場(chǎng)景,在異步調(diào)用時(shí)還是使用自定義線程池比較好。

  • 針對(duì)開(kāi)篇提到的兩個(gè)顯而易見(jiàn)的好處,此處新增一條:

    • 可以根據(jù)我們的服務(wù)器性能,通過(guò)池的管理更好的規(guī)劃我們的線程數(shù)。
    • 可以對(duì)我們使用的線程自定義名稱,這里也是阿里java開(kāi)發(fā)規(guī)范所提到的。
    • 不會(huì)因?yàn)樽枞麑?dǎo)致使用共享線程池的其他線程阻塞甚至異常。

我們自定義下面的線程池:

/**
 * @description: 全局通用線程池
 * @author:weirx
 * @date:2021/9/9 18:09
 * @version:3.0
 */
@Slf4j
public class GlobalThreadPool {

    /**
     * 核心線程數(shù)
     */
    public final static int CORE_POOL_SIZE = 10;

    /**
     * 最大線程數(shù)
     */
    public final static int MAX_NUM_POOL_SIZE = 20;

    /**
     * 任務(wù)隊(duì)列大小
     */
    public final static int BLOCKING_QUEUE_SIZE = 30;

    /**
     * 線程池實(shí)例
     */
    private final static ThreadPoolExecutor instance = getInstance();


    /**
     * description: 初始化線程池
     *
     * @return: java.util.concurrent.ThreadPoolExecutor
     * @author: weirx
     * @time: 2021/9/10 9:49
     */
    private synchronized static ThreadPoolExecutor getInstance() {
        // 生成線程池
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                CORE_POOL_SIZE,
                MAX_NUM_POOL_SIZE,
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(BLOCKING_QUEUE_SIZE),
                new NamedThreadFactory("Thread-wjbgn-", false));
        return executor;
    }

    private GlobalThreadPool() {
    }

    public static ThreadPoolExecutor getExecutor() {
        return instance;
    }
}

調(diào)用:

    public static void main(String[] args) {
        // 創(chuàng)建10個(gè)任務(wù),每個(gè)任務(wù)阻塞10秒
        for (int i = 0; i < 10; i++) {
            CompletableFuture.runAsync(() -> {
                try {
                    Thread.sleep(10000);
                    System.out.println(new Date() + ":" + Thread.currentThread().getName());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            },GlobalThreadPool.getExecutor());
        }

        try {
            Thread.sleep(30000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

輸出我們指定線程名稱的線程:

Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-1
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-10
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-2
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-9
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-5
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-6
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-3
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-7
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-8
Mon Sep 13 11:32:35 CST 2021:Thread-Inbox-Model-4

三、題外話,動(dòng)態(tài)線程池

3.1 什么是動(dòng)態(tài)線程池?

在我們使用線程池的時(shí)候,是否有的時(shí)候很糾結(jié),到底設(shè)置多大的線程池參數(shù)是最合適的呢?如果不夠用了怎么辦,要改代碼重新部署嗎?

其實(shí)是不需要的,記得當(dāng)初看過(guò)美團(tuán)的一篇文章,真的讓人茅塞頓開(kāi)啊,動(dòng)態(tài)線程池。

ThreadPoolExecutor這個(gè)類(lèi)其實(shí)是提供對(duì)于線程池的屬性進(jìn)行修改的,支持我們動(dòng)態(tài)修改以下的屬性:

從上至下分別是:

  • 線程工廠(用于指定線程名稱)
  • 核心線程數(shù)
  • 最大線程數(shù)
  • 活躍時(shí)間
  • 拒絕策略。

在美團(tuán)的文章當(dāng)中呢,是監(jiān)控服務(wù)器線程的使用率,當(dāng)達(dá)到閾值就進(jìn)行告警,然后通過(guò)配置中心去動(dòng)態(tài)修改這些數(shù)值。

我們也可以這么做,使用@RefreshScope加nacos就可以實(shí)現(xiàn)了。

3.2 實(shí)踐

我寫(xiě)了一個(gè)定時(shí)任務(wù),監(jiān)控當(dāng)前服務(wù)的線程使用率,小了就擴(kuò)容,一段時(shí)間后占用率下降,就恢復(fù)初始值。

其實(shí)還有很多地方需要改進(jìn)的,請(qǐng)大家多提意見(jiàn),監(jiān)控的是文章前面的線程池GlobalThreadPool,下面調(diào)度任務(wù)的代碼:

/**
 * @description: 全局線程池守護(hù)進(jìn)程
 * @author:weirx
 * @date:2021/9/10 16:32
 * @version:3.0
 */
@Slf4j
@Component
public class DaemonThreadTask {

    /**
     * 服務(wù)支持最大線程數(shù)
     */
    public final static int SERVER_MAX_SIZE = 50;

    /**
     * 最大閾值Maximum threshold,百分比
     */
    private final static int MAXIMUM_THRESHOLD = 8;

    /**
     * 每次遞增最大線程數(shù)
     */
    private final static int INCREMENTAL_MAX_NUM = 10;

    /**
     * 每次遞增核心線程數(shù)
     */
    private final static int INCREMENTAL_CORE_NUM = 5;

    /**
     * 當(dāng)前線程數(shù)
     */
    private static int currentSize = GlobalThreadPool.MAX_NUM_POOL_SIZE;

    /**
     * 當(dāng)前核心線程數(shù)
     */
    private static int currentCoreSize = GlobalThreadPool.CORE_POOL_SIZE;

    @Scheduled(cron = "0 */5 * * * ?")
    public static void execute() {
        threadMonitor();
    }


    /**
     * description: 動(dòng)態(tài)監(jiān)控并設(shè)置線程參數(shù)
     *
     * @return: void
     * @author: weirx
     * @time: 2021/9/10 13:20
     */
    private static void threadMonitor() {
        ThreadPoolExecutor instance = GlobalThreadPool.getExecutor();
        int activeCount = instance.getActiveCount();
        int size = instance.getQueue().size();
        log.info("GlobalThreadPool: the active thread count is {}", activeCount);
        // 線程數(shù)不足,增加線程
        if (activeCount > GlobalThreadPool.MAX_NUM_POOL_SIZE % MAXIMUM_THRESHOLD
                && size >= GlobalThreadPool.BLOCKING_QUEUE_SIZE) {
            currentSize = currentSize + INCREMENTAL_MAX_NUM;
            currentCoreSize = currentCoreSize + INCREMENTAL_CORE_NUM;
            //當(dāng)前設(shè)置最大線程數(shù)小于服務(wù)最大支持線程數(shù)才可以繼續(xù)增加線程
            if (currentSize <= SERVER_MAX_SIZE) {
                instance.setMaximumPoolSize(currentSize);
                instance.setCorePoolSize(currentCoreSize);
                log.info("this max thread size is {}", currentSize);
            } else {
                log.info("current size is more than server max size, can not add");
            }
        }
        // 線程數(shù)足夠,降低線程數(shù),當(dāng)前活躍數(shù)小于默認(rèn)核心線程數(shù)
        if (activeCount < GlobalThreadPool.MAX_NUM_POOL_SIZE
                && size == 0
                && currentSize > GlobalThreadPool.MAX_NUM_POOL_SIZE) {
            currentSize = GlobalThreadPool.MAX_NUM_POOL_SIZE;
            currentCoreSize = GlobalThreadPool.CORE_POOL_SIZE;
            instance.setMaximumPoolSize(currentSize);
            instance.setCorePoolSize(currentCoreSize);
        }
    }
}

3.3 動(dòng)態(tài)線程池有什么意義?

有的朋友其實(shí)問(wèn)過(guò)我,我直接把線程池設(shè)置大一點(diǎn)不就好了,這種動(dòng)態(tài)線程池有什么意義呢?

其實(shí)這是一個(gè)好問(wèn)題。在以前的傳統(tǒng)軟件當(dāng)中,單機(jī)部署,硬件部署,確實(shí),我們能使用的線程數(shù)取決于服務(wù)器的核心線程數(shù),而且基本沒(méi)有其他服務(wù)來(lái)爭(zhēng)搶這些線程。

但是現(xiàn)在是容器的時(shí)代,云原生的時(shí)代。

多個(gè)容器部署在一個(gè)宿主機(jī)上,那么當(dāng)高峰期的時(shí)候,某個(gè)容器就需要占用大量的cpu資源,如果所有的容器都將大部分資源占據(jù),那么這個(gè)容器必然面臨阻塞甚至癱瘓的風(fēng)險(xiǎn)。

當(dāng)高峰期過(guò)了,釋放這部分資源可以被釋放掉,用于其他需要的容器。。

再結(jié)合到目前的云服務(wù)器節(jié)點(diǎn)擴(kuò)容,都是需要?jiǎng)討B(tài)擴(kuò)縮容的的,和線程相似,在滿足高可用的情況下,盡量的節(jié)約成本。

總結(jié)

到此這篇關(guān)于java8異步調(diào)用如何使用的文章就介紹到這了,更多相關(guān)java8異步調(diào)用使用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!

相關(guān)文章

最新評(píng)論