Springboot中使用線程池的三種方式

来源:https://www.cnblogs.com/LoveBB/archive/2023/09/19/17713614.html
-Advertisement-
Play Games

前言 多線程是每個程式員的噩夢,用得好可以提升效率很爽，用得不好就是埋汰的火葬場。 這裡不深入介紹，主要是講解一些標準用法，熟讀唐詩三百首，不會作詩也會吟。 這裡就介紹一下springboot中的多線程的使用，使用線程連接池去非同步執行業務方法。 由於代碼中包含詳細註釋,也為了保持文章的整潔性，我就不 ...


前言

  • 多線程是每個程式員的噩夢,用得好可以提升效率很爽，用得不好就是埋汰的火葬場。
  • 這裡不深入介紹，主要是講解一些標準用法，熟讀唐詩三百首，不會作詩也會吟。
  • 這裡就介紹一下springboot中的多線程的使用，使用線程連接池去非同步執行業務方法。
  • 由於代碼中包含詳細註釋,也為了保持文章的整潔性，我就不過多的做文字描述了。

VisiableThreadPoolTaskExecutor 編寫

  • new VisiableThreadPoolTaskExecutor() 方式創建線程池, 返回值是 Executor
點擊查看代碼 

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.util.concurrent.ListenableFuture;

import java.util.concurrent.Callable;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author love ice
 * @create 2023-09-19 0:17
 */
@Slf4j
public class VisiableThreadPoolTaskExecutor extends ThreadPoolTaskExecutor {
    @Override
    public void execute(Runnable task){
        showThreadPoolInfo("execute一個參數的方法執行");
    }

    @Override
    public void execute(Runnable task, long startTimeout){
        showThreadPoolInfo("execute兩個參數的方法執行");
    }

    @Override
    public Future<?> submit(Runnable task){
        showThreadPoolInfo("submit Runnable task 入參方法執行");
        return super.submit(task);
    }

    @Override
    public <T> Future<T> submit(Callable<T> task){
        showThreadPoolInfo("submit Callable<T> task 入參方法執行");
        return super.submit(task);
    }

    @Override
    public ListenableFuture<?> submitListenable(Runnable task){
        showThreadPoolInfo("submitListenable(Runnable task) 方法執行");
        return super.submitListenable(task);
    }

    @Override
    public <T>ListenableFuture<T> submitListenable(Callable<T> task){
        showThreadPoolInfo("submitListenable(Callable<T> task) 方法執行");
        return super.submitListenable(task);
    }

    private void showThreadPoolInfo(String prefix){
        ThreadPoolExecutor threadPoolExecutor = getThreadPoolExecutor();
        log.info("{}, {}, taskCount[{}], completedTaskCount[{}], activeCount[{}], queueSize[{}]",
                this.getThreadNamePrefix(), prefix, threadPoolExecutor.getTaskCount(),
                threadPoolExecutor.getCompletedTaskCount(), threadPoolExecutor.getActiveCount(),
                threadPoolExecutor.getQueue().size());
    }
}

ThreadExceptionLogHandler 編寫

  • 主要用於線程池出現異常時的捕獲
點擊查看代碼 
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author love ice
 * @create 2023-09-19 0:13
 */
@Slf4j
@Component
public class ThreadExceptionLogHandler implements Thread.UncaughtExceptionHandler {

    @Override
    public void uncaughtException(Thread t, Throwable e) {
        log.error("[{}]線程池異常,異常信息為:{}",t.getName(),e.getMessage(),e);
    }
}

ExecutorConfig 編寫

  • 核心配置類
點擊查看代碼 

import com.test.redis.Infrastructure.handler.ThreadExceptionLogHandler;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 線程池配置
 *
 * @author love ice
 * @create 2023-09-19 0:09
 */
@Configuration
@EnableAsync
public class ExecutorConfig {
    @Value("${thread.pool.coreSize:50}")
    private int coreSize;
    @Value("${thread.pool.maxSize:50}")
    private int maxSize;
    @Value("${thread.pool.queueSize:9999}")
    private int queueSize;
    @Value("${thread.pool.threadNamePrefix:thread-name}")
    private String threadNamePrefix;
    @Value("${thread.pool.keepAlive:60}")
    private int keepAlive;
    @Autowired
    private ThreadExceptionLogHandler threadExceptionLogHandler;

    /**
     * 方式一: new VisiableThreadPoolTaskExecutor() 方式創建線程池,返回值是 Executor
     * 適用於 @Async("asyncServiceExecutor") 註解
     * 也可以
     *  @Autowired
     *  private Executor asyncServiceExecutor;
     *
     * @return Executor
     */
    @Bean
    public Executor asyncServiceExecutor() {
        VisiableThreadPoolTaskExecutor executor = new VisiableThreadPoolTaskExecutor();
        // 配置核心線程數 50
        executor.setCorePoolSize(coreSize);
        // 配置最大線程數 50
        executor.setMaxPoolSize(maxSize);
        // 配置隊列大小 9999
        executor.setQueueCapacity(queueSize);
        // 配置線程池中的線程名稱首碼 模塊-功能-作用
        executor.setThreadNamePrefix(threadNamePrefix);
        // rejection-policy：當pool已經達到max size的時候，如何處理新任務
        // CALLER_RUNS：不在新線程中執行任務，而是有調用者所在的線程來執行
        // 線程池無法接受新的任務並且隊列已滿時，如果有新的任務提交給線程池，而線程池已經達到了最大容量限制，那麼這個任務不會被丟棄，而是由調用該任務的線程來執行。
        // 這樣可以避免任務被直接丟棄，並讓調用者自己執行任務以減輕任務提交頻率。
        // 這個拒絕策略可能會導致任務提交者的線程執行任務，這可能會對調用者的性能產生一些影響，因為調用者線程需要等待任務執行完成才能繼續進行其他操作。
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        // 線程空閑後的最大存活時間 60
        executor.setKeepAliveSeconds(keepAlive);
        // 執行初始化
        executor.initialize();
        return executor;
    }

    /**
     * 方式二: new ThreadPoolExecutor() 方式創建線程池
     * 適用於:
     * @Autowired
     * private ExecutorService fbWorkerPool;
     * @return ExecutorService
     */
    @Bean
    public ExecutorService workerPool() {
        return new ThreadPoolExecutor(coreSize, maxSize, keepAlive, TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<>(20000),
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);
                    @Override
                    public Thread newThread(Runnable runnable) {
                        Thread thread = new Thread(runnable, threadNamePrefix + threadNumber.getAndIncrement());
                        thread.setUncaughtExceptionHandler(threadExceptionLogHandler);
                        return thread;
                    }
                });
    }
}

ExecutorController 編寫

  • 演示demo，三種不同的用法, 足以涵蓋大部分場景
點擊查看代碼 

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.stream.Collectors;

/**
 * 這裡是demo演示、把業務寫在 controller 了,一般開發都是在 service 層實現的。
 *
 * @author love ice
 * @create 2023-09-19 0:59
 */
@RestController
@RequestMapping("/executor")
public class ExecutorController {


    /**
     * demo1: 使用非同步註解 @Async("asyncServiceExecutor") 執行方法,適用於沒有返回值的情況下
     */
    public void asyncDemo1() {
        // 假設這是從資料庫查詢出來的數據
        List<String> nameList = new ArrayList<>(Arrays.asList("張三", "李四", "王五"));
        // 把 nameList 進行切分
        int j = 0, size = nameList.size(), batchSize = 10;
        List<List<String>> list = new ArrayList<>();
        while (j < size) {
            List<String> batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());
            list.add(batchList);
            j += batchSize;
        }

        // 先把 list 切分成小份數據,在使用 @Async(),非同步處理數據
        list.stream().parallel().forEach(this::asynchronousAuthorization1);
    }

    /**
     * 非同步註解處理業務邏輯,實際業務開發,需要提取到 Service 層,否則會報錯。
     *
     * @param paramList 入參
     */
    @Async("asyncServiceExecutor")
    public void asynchronousAuthorization1(List<String> paramList) {
        paramList.forEach(System.out::println);
        System.out.println("非同步執行 paramList 業務邏輯");
    }

//================================分隔符======================

    @Autowired
    private ExecutorService workerPool;

    /**
     * demo2: workerPool.execute() 實現非同步邏輯。適用於沒有返回值的情況下
     */
    public void asyncDemo2() {
        // 假設這是從資料庫查詢出來的數據
        List<String> nameList = new ArrayList<>(Arrays.asList("張三", "李四", "王五"));
        // 把 nameList 進行切分
        int j = 0, size = nameList.size(), batchSize = 10;
        List<List<String>> list = new ArrayList<>();
        while (j < size) {
            List<String> batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());
            list.add(batchList);
            j += batchSize;
        }

        // 將 list 切分成小份數據,workerPool.execute(),非同步處理數據
        list.stream().parallel().forEach(paramList->{
            workerPool.execute(()->asynchronousAuthorization2(paramList));
        });
    }

    public void asynchronousAuthorization2(List<String> paramList) {
        paramList.forEach(System.out::println);
        System.out.println("非同步執行 paramList 業務邏輯");
    }


//================================分隔符======================


    /**
     * demo3: futures.add() 實現非同步邏輯。適用於有返回值的情況下
     */
    public void asyncDemo3() {
        // 假設這是從資料庫查詢出來的數據
        List<String> nameList = new ArrayList<>(Arrays.asList("張三", "李四", "王五"));
        // 把 nameList 進行切分
        int j = 0, size = nameList.size(), batchSize = 10;
        List<List<String>> list = new ArrayList<>();
        while (j < size) {
            List<String> batchList = nameList.stream().skip(j).limit(Math.min(j + batchSize, size) - j).collect(Collectors.toList());
            list.add(batchList);
            j += batchSize;
        }

        List<CompletableFuture<String>> futures = new ArrayList<>();
        // 將 list 切分成小份數據,futures.add(),非同步處理數據,有返回值的情況下
        list.forEach(paramList->{
            // CompletableFuture.supplyAsync() 該任務會在一個新的線程中執行，並返回一個結果
            // 通過futures.add(...)將這個非同步任務添加到futures列表中。這樣可以方便後續對多個非同步任務進行管理和處理
            futures.add(CompletableFuture.supplyAsync(() -> {
                asynchronousAuthorization3(paramList);
               return "預設值";
            }, workerPool));

            // 防止太快,讓它休眠一下
            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        //new CompletableFuture[0] 創建了一個初始長度為 0 的 CompletableFuture 數組，作為目標數組。然後，futures.toArray(new CompletableFuture[0]) 將 futures 列表中的元素複製到目標數組中，並返回結果數組。
        CompletableFuture<String>[] futuresArray = futures.toArray(new CompletableFuture[0]);
        // 通過將多個非同步任務添加到futures列表中，我們可以使用CompletableFuture提供的方法來對這些非同步任務進行組合、等待和處理。
        // 例如使用CompletableFuture.allOf(...)等待所有任務完成，或者使用CompletableFuture.join()獲取單個任務的結果等。
        CompletableFuture.allOf(futures.toArray(futuresArray)).join();

        // 獲取每個任務的結果或處理異常
        List<String> results = new ArrayList<>();
        for (CompletableFuture<String> future :futuresArray) {
            // 處理任務的異常
            future.exceptionally(ex -> {
                System.out.println("Task encountered an exception: " + ex.getMessage());
                return "0"; // 返回預設值或者做其他補償操作
            });

            // 獲取任務結果
            String result = future.join();
            results.add(result);
        }
        // 所有任務已完成，可以進行下一步操作
    }

    public void asynchronousAuthorization3(List<String> paramList) {
        paramList.forEach(System.out::println);
        System.out.println("非同步執行 paramList 業務邏輯");
    }
}
作者：天下沒有收費的bug 出處：https://www.cnblogs.com/LoveBB/ 本文版權歸作者和博客園共有，歡迎轉載，但未經作者同意必須在文章頁面給出原文鏈接，否則保留追究法律責任的權利。
您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • Ubuntu20.04安裝Postgres主從備份
    Ubuntu20.04安裝Postgres主從備份 一.查看可安裝的Postgres包 #列出相關的軟體包，這裡安裝的是14版本 apt list | grep -w postgresql-14 | tail -1 #下載Postgres apt install -y postgresql-14/f ...
  • 萬字長文教你實現華為雲IoT+OpenHarmony智能家居開發
    基於OpenHarmony和華為雲平臺打造的智能家居設備，分別為智能門鎖，儲物精靈 NFC版，儲物精靈Pro版三個設備。 ...
  • iOS16新特性：實時活動-在鎖屏界面實時更新APP消息
    之前在 《iOS16新特性:靈動島適配開發與到家業務場景結合的探索實踐》 里介紹了iOS16新的特性：實時更新（Live Activity）中靈動島的適配流程，但其實除了靈動島的展示樣式，Live Activity還有一種非常實用的應用場景，那就是鎖屏界面實時狀態更新： ...
  • 前端遠程調試方案 Chii 的使用經驗分享
    前端遠程調試方案 Chii 的使用經驗分享 Chii 是與 weinre 一樣的遠程調試工具 ，主要是將 web inspector 替換為最新的 chrome devtools frontend 監控列表頁面可以看到網站的標題鏈接，IP，useragent，可以快速定位調試頁面，監控頁信息完善，支 ...
  • 記錄--JavaScript 令人驚訝的一點：對於空數組every()方法返回true
    這裡給大家分享我在網上總結出來的一些知識，希望對大家有所幫助 JavaScript 語言的內核足夠大，導致我們很容易誤解它的某些部分是如何工作的。我最近重構了一些使用 every ()方法的代碼，並且發現我並不真正理解every()的邏輯。在我看來，我認為回調函數必須被調用並返回 true的時候ev ...
  • 前端模塊化
    今天在維護優化公司中台項目時，發現路由的文件配置非常多非常亂，只要只中大型項目，都會進入很多的路由頁面，規範一點的公司還會吧路由進行模塊化導入，但是依然存在很多文件夾的和手動導入的問題。 於是我想到了我之前使用vuex時進行的模塊化自動導入js文件，能不能使用到自動導入.vue文件中去，答案是可以！ ...
  • ES 2023新特性速解
    ES 2023新特性速解 一、新增數組方法 操作數組的方法 Array.prototype.toSorted(compareFn) //返回一個新數組，其中元素按升序排序，而不改變原始數組。 Array.prototype.toReversed() //返回一個新數組，該數組的元素順序被反轉，但不改 ...
  • 1、spring_helloworld
    這是一個講解DDD落地的文章系列，作者是《實現領域驅動設計》的譯者滕雲。本文章系列以一個真實的並已成功上線的軟體項目——碼如雲（https://www.mryqr.com）為例，系統性地講解DDD在落地實施過程中的各種典型實踐，以及在面臨實際業務場景時的諸多取捨。 本系列包含以下文章： DDD入門 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類