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

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

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

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

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

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

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

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

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

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

   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í)別并沒有給出默認(rèn)值

static final int commonParallelism;

通過找到這個(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;

下面我們寫個(gè)main方法測試一下，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密集型，簡單說處理時(shí)間占用長，將導(dǎo)致其他使用共享線程池的任務(wù)阻塞，造成系統(tǒng)性能下降甚至異常。甚至當(dāng)一部分調(diào)用接口時(shí)，如果接口超時(shí)，那么也會(huì)阻塞與超時(shí)時(shí)長相同的時(shí)間；實(shí)際在計(jì)算密集的場景下使用是能提高性能的。

二、使用自定義的線程池

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

  • 針對(duì)開篇提到的兩個(gè)顯而易見的好處，此處新增一條：

    • 可以根據(jù)我們的服務(wù)器性能，通過池的管理更好的規(guī)劃我們的線程數(shù)。
    • 可以對(duì)我們使用的線程自定義名稱，這里也是阿里java開發(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)初看過美團(tuán)的一篇文章，真的讓人茅塞頓開啊，動(dòng)態(tài)線程池。

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

從上至下分別是：

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

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

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

3.2 實(shí)踐

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

其實(shí)還有很多地方需要改進(jì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í)問過我，我直接把線程池設(shè)置大一點(diǎn)不就好了，這種動(dòng)態(tài)線程池有什么意義呢？

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

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

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

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

再結(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)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

相關(guān)文章

最新評(píng)論