2. 程式人生 > 實用技巧 >執行緒池工具類幾種實現

執行緒池工具類幾種實現

阿新 • • 發佈：2020-11-18

一 執行緒池工具類
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**

  • @Description 執行緒池工具類
    */
    public class ThreadPoolUtil {

    /**

    • 核心執行緒數，會一直存活，即使沒有任務，執行緒池也會維護執行緒的最少數量
      /
      private static final int SIZE_CORE_POOL = 5;
      /
      *
    • 執行緒池維護執行緒的最大數量
      /
      private static final int SIZE_MAX_POOL = 10;
      /      *
    • 執行緒池維護執行緒所允許的空閒時間
      /
      private static final long ALIVE_TIME = 2000;
      /      *
    • 執行緒緩衝佇列
      */
      private static BlockingQueue bqueue = new ArrayBlockingQueue(100);
      private static ThreadPoolExecutor pool = new ThreadPoolExecutor(SIZE_CORE_POOL, SIZE_MAX_POOL, ALIVE_TIME, TimeUnit.MILLISECONDS, bqueue, new ThreadPoolExecutor.CallerRunsPolicy());

    static {
    pool.prestartAllCoreThreads();
    }

    public static ThreadPoolExecutor getPool() {
    return pool;
    }
    }

測試類
import com.dashuai.cloud.consulconsumer.util.ThreadPoolUtil;

public class TestUtil {
public static void main(String[] args) {
ThreadPoolUtil.getPool().execute(new Runnable() {
@Override
public void run() {
System.out.println("執行緒池呼叫");
}
});
}
}

二 執行緒池支援多執行緒返回結果
import org.springframework.stereotype.Service;

/**

  • ClassName:CommenThreadPoolUtil
  • Function:執行緒池公共入口處理類.

*/
@Service
public class CommonThreadPoolUtil {

// 核心執行緒數(預設初始化為10)
private int cacheCorePoolSize = 8;

// 核心執行緒控制的最大數目
private int maxCorePoolSize = 160;

// 佇列等待執行緒數閾值
private int blockingQueueWaitSize = 16;

// 核心執行緒數自動調整的增量幅度
private int incrementCorePoolSize = 4;

// 初始化執行緒物件ThreadLocal,重寫initialValue()，保證ThreadLocal首次執行get方法時不會null異常
private ThreadLocal<List<Future<?>>> threadlocal = new ThreadLocal<List<Future<?>>>() {

    protected List<Future<?>> initialValue() {

        return new ArrayList<Future<?>>();
    }
};

// 初始化執行緒池
private MyselfThreadPoolExecutor ThreadPool = new MyselfThreadPoolExecutor(cacheCorePoolSize, cacheCorePoolSize, 0L,
        TimeUnit.MILLISECONDS, new LinkedBlockingDeque<Runnable>());

/**
 *
 * dealTask:(執行緒池執行操作-包含每個程序返回結果). <br/>
 * 1、運用場景：例如，需要同時校驗很多不同的邏輯，依賴於獲取校驗結果響應給使用者； 2、具體實現java類：implements
 * 的Callable介面，重寫call方法即可，支援返回值
 *
 * @author
 * @param callable
 * @return
 */
public Map<String, Object> dealTask(Callable<?> callable) {

    try {
        // 動態更改核心執行緒數大小
        dynamicTuningPoolSize();
        // 執行執行緒業務邏輯及獲取返回結果
        Future<?> result = ThreadPool.submit(callable);
        // 獲取當前程序的區域性變數
        List<Future<?>> threadLocalResult = threadlocal.get();
        // 疊加主程序對應的多個程序處理結果
        threadLocalResult.add(result);
        // 設定最新的threadLocal變數到當前主程序
        threadlocal.set(threadLocalResult);
    } catch (Exception e) {
        e.printStackTrace();
        return errorResp("執行緒池發生異常-Future", null);
    }
    return successResp(null);
}

/**
 *
 * dealTask:(執行緒池執行操作-不包含每個程序返回結果). <br/>
 * 1、運用場景：例如，不依賴於響應給使用者執行結果的業務邏輯 ； 2、具體實現java類：implements
 * 的Runnable介面，重寫run方法，沒有返回值
 *
 * @author
 * @param runnable
 * @return
 */
public Map<String, Object> dealTask(Runnable runnable) {

    try {
        // 動態更改核心執行緒數大小
        dynamicTuningPoolSize();
        // 執行執行緒業務邏輯
        ThreadPool.execute(runnable);
    } catch (Exception e) {
        e.printStackTrace();
        return errorResp("執行緒池發生異常", null);
    }
    return successResp(null);
}

/**
 * obtainTaskFuture:(獲取執行緒池執行結果：此為阻塞執行緒，即所有執行緒都執行完成才能獲取結果，故應將執行時間稍長的業務邏輯先執行，
 * 減少等待時間). <br/>
 * 此方法只能呼叫一次，即呼叫之後清除ThreadLocal變數，以便於同一程序再次呼叫執行緒池獲取最新的執行結果以及釋放記憶體， 防止記憶體洩露
 *
 * @author
 * @return
 */
public Map<String, Object> obtainTaskFuture() {

    List<Future<?>> threadLocalResult = null;
    try {
        // 獲取當前程序變數
        threadLocalResult = threadlocal.get();
        if (threadLocalResult == null || threadLocalResult.size() == 0) {
            return errorResp("獲取執行緒池執行結果為空", null);
        } else {
            return successResp(threadLocalResult);
        }
    } catch (Exception e) {
        return errorResp("獲取執行緒池執行結果發生異常:" + e.getMessage(), null);
    } finally {
        // 1、釋放記憶體；2、防止主程序再次呼叫執行緒池方法時對結果互有影響。
        threadlocal.remove();
    }

}

/**
 *
 * dynamicTuningPoolSize:(動態改變核心執行緒數). <br/>
 *
 * @author
 * @return
 */
private void dynamicTuningPoolSize() {

    // 佇列等待任務數（此為近似值，故採用>=判斷）
    int queueSize = ThreadPool.getQueueSize();
    // 動態更改核心執行緒數大小
    if (queueSize >= blockingQueueWaitSize) {
        // 核心執行緒數小於設定的最大執行緒數才會自動擴充套件執行緒數
        if (cacheCorePoolSize <= maxCorePoolSize) {
            // 原有核心執行緒數
            int corePoolSize = ThreadPool.getCorePoolSize();
            // 將要累積的核心執行緒數
            int currentcorePoolSize = corePoolSize + incrementCorePoolSize;
            ThreadPool.setCorePoolSize(currentcorePoolSize);
            ThreadPool.setMaximumPoolSize(currentcorePoolSize);
            cacheCorePoolSize = currentcorePoolSize;
            System.out.println("動態改變執行緒池大小====原核心執行緒池數目為：" + corePoolSize + ";現累加為：" + currentcorePoolSize);
        } else {
            System.out.println("動態改變執行緒池大小====核心執行緒池數目已累加為：" + cacheCorePoolSize + "；不會繼續無限增加");
        }
    }
}

/**
 * 獲取核心執行緒數 getCacheCorePoolSize:(). <br/>
 *
 * @author
 * @return
 */
public int getCacheCorePoolSize() {

    return ThreadPool.getCorePoolSize();
}

/**
 * 設定核心執行緒數 setCacheCorePoolSize:(). <br/>
 *
 * @author
 * @param cacheCorePoolSize
 */
public void setCacheCorePoolSize(int cacheCorePoolSize) {

    ThreadPool.setCorePoolSize(cacheCorePoolSize);
    ThreadPool.setMaximumPoolSize(cacheCorePoolSize);
    this.cacheCorePoolSize = cacheCorePoolSize;
}

/**
 *
 * successResp:(正確響應資訊). <br/>
 *
 * @author
 * @param data
 * @return
 */
private Map<String, Object> successResp(Object data) {

    Map<String, Object> result = new HashMap<String, Object>();
    result.put("status", "0");
    result.put("data", data);
    return result;

}

/**
 *
 * errorResp:(錯誤響應資訊). <br/>
 *
 * @author
 * @param errorMsg
 * @param data
 * @return
 */
public Map<String, Object> errorResp(String errorMsg, Object data) {

    Map<String, Object> result = new HashMap<String, Object>();
    result.put("status", "1");
    result.put("msg", errorMsg);
    result.put("data", data);
    return result;

}

}

建立執行緒池類

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class MyselfThreadPoolExecutor extends ThreadPoolExecutor {

// 初始化父類建構函式及startTime
public MyselfThreadPoolExecutor(int corePoolSize, int maximumPoolSize,
		long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue) {

	super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
}

// 按過去執行已提交任務的順序發起一個有序的關閉，但是不接受新任務(已執行的任務不會停止)
@Override
public void shutdown() {

	super.shutdown();

}

// 嘗試停止所有的活動執行任務、暫停等待任務的處理，並返回等待執行的任務列表。在從此方法返回的任務佇列中排空（移除）這些任務。並不保證能夠停止正在處理的活動執行任務，但是會盡力嘗試。
@Override
public List<Runnable> shutdownNow() {

	return super.shutdownNow();

}

// 在執行給定執行緒中的給定 Runnable 之前呼叫的方法.可用於重新初始化ThreadLocals或者執行日誌記錄。
@Override
protected void beforeExecute(Thread t, Runnable r) {

	super.beforeExecute(t, r);
}

// 基於完成執行給定 Runnable 所呼叫的方法
@Override
protected void afterExecute(Runnable r, Throwable t) {

	super.afterExecute(r, t);

	try {
		// Future<?> result = (Future<?>) r;
		// "任務結果：" result.get();
	} catch (Exception e) {
	}
}

/**
 * 
 * getQueueSize:(已執行的任務數). <br/>
 *
 * @author
 * @return
 */
@Override
public long getCompletedTaskCount() {

	return super.getCompletedTaskCount();
}

/**
 * 
 * getQueueSize:(正在執行的任務數). <br/>
 *
 * @author
 * @return
 */
@Override
public int getActiveCount() {

	return super.getActiveCount();
}

/**
 * 
 * getQueueSize:(佇列等待任務數). <br/>
 *
 * @author
 * @return
 */
public int getQueueSize() {

	return getQueue().size();
}

}
測試類
public class TestUtil {
public static void main(String[] args) {

    CommonThreadPoolUtil poolUtil = new CommonThreadPoolUtil();
    poolUtil.dealTask(new Runnable() {
        @Override
        public void run() {
            System.out.println("執行緒池呼叫");
        }
    });
}

}

三 jdk1.5之後提供工具類 Executors
工具類Executors面提供了一些靜態工廠方法，生成一些常用的執行緒池，如下所示：

  • newCachedThreadPool：建立一個可快取的執行緒池。如果執行緒池的大小超過了處理任務所需要的執行緒，那麼就會回收部分空閒（60秒不執行任務）的執行緒，當任務數增加時，此執行緒池又可以智慧的新增新執行緒來處理任務。此執行緒池不會對執行緒池大小做限制（Interger. MAX_VALUE），執行緒池大小完全依賴於作業系統（或者說JVM）能夠建立的最大執行緒大小。

  • newFixedThreadPool：建立固定大小的執行緒池。每次提交一個任務就建立一個執行緒，直到執行緒達到執行緒池的最大大小。執行緒池的大小一旦達到最大值就會保持不變，如果某個執行緒因為執行異常而結束，那麼執行緒池會補充一個新執行緒。

  • newSingleThreadExecutor：建立一個單執行緒的執行緒池。這個執行緒池只有一個執行緒在工作，也就是相當於單執行緒序列執行所有任務。如果這個唯一的執行緒因為異常結束，那麼會有一個新的執行緒來替代它。此執行緒池保證所有任務的執行順序按照任務的提交順序執行。

  • newScheduledThreadPool：建立一個大小無限的執行緒池。此執行緒池支援定時以及週期性執行任務的需求。

總結：除了newScheduledThreadPool的內部實現特殊一點之外，其它執行緒池內部都是基於 ThreadPoolExecutor 類（Executor的子類）實現的。
實現：
ScheduledExecutorService scheduExec = Executors.newScheduledThreadPool(10);
scheduExec.schedule(new Runnable() {

  @SuppressWarnings("static-access")
  @Override
  public void run() {
  }

}, 5, TimeUnit.MINUTES);
週期性定時任務5分鐘後執行

相關推薦

執行緒池工具類幾種實現

一 執行緒池工具類 import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue;

java ThreadPool執行緒池的使用,執行緒池工具類用法說明

實際上java已經提供執行緒池的實現 ExecutorService。 為了更方便的使用和管理。這裡提供一個執行緒池工具類，方便大家的使用。

ThreadPoolExecutor執行緒池工具類

技術標籤：併發工具java執行緒多執行緒併發程式設計threadjava多執行緒 import lombok.AccessLevel;

JDK1.8 建立執行緒池有哪幾種方式？

newFixedThreadPool 定長執行緒池，每當提交一個任務就建立一個執行緒，直到達到執行緒池的最大數量，這時執行緒數量不再變化，當執行緒發生錯誤結束時，執行緒池會補充一個新的執行緒

Java執行緒併發工具類CountDownLatch原理及用法

一、CountDownLatch 【1】CountDownLatch是什麼？ CountDownLatch，英文翻譯為倒計時鎖存器，是一個同步輔助類，在完成一組正在其他執行緒中執行的操作之前，它允許一個或

java 定時器執行緒池（ScheduledThreadPoolExecutor）的實現

前言 定時器執行緒池提供了定時執行任務的能力，即可以延遲執行，可以週期性執行。但定時器執行緒池也還是執行緒池，最底層實現還是ThreadPoolExecutor，可以參考我的另外一篇文章多執行緒–精通ThreadPoolExecutor

執行緒相關工具類

public class Threads {/*** sleep等待,單位為毫秒,忽略InterruptedException.*/public static void sleep(long millis) {try {Thread.sleep(millis);} catch (InterruptedException e) {// Ignore.return;}}/*** sl

多執行緒併發工具類01-CountDownLatch 執行緒工具類

     CountDownLatch:概念是，允許一個或多個執行緒等待其他執行緒完成操作；

自定義非同步執行緒池工具，用於執行非同步方法

此方法經歷百萬資料量線上實踐驗證過 import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor;

JVM優先順序執行緒池做任務佇列的實現方法

前言 我們都知道 web 服務的工作大多是接受 http 請求，並返回處理後的結果。伺服器接受的每一個請求又可以看是一個任務。一般而言這些請求任務會根據請求的先後有序處理，如果請求任務的處理比較耗時，往往就需要排

【併發程式設計】6.執行緒控制工具類

1.Future、Callable、FutureTask 執行緒池提供的submit方法 <T> Future<T> submit(Callable<T> task); //Callablecall方法具有返回值，Future物件可以通過呼叫其get()方法來獲 取任務的執行結果。

java執行緒池工具

ExecutorServiceFactory package com.nblh.office.threadpool; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ExecutorService;

執行緒池原理及C語言實現執行緒池

備註：該執行緒池原始碼參考自傳直播客培訓視訊配套資料； 原始碼：https://pan.baidu.com/s/1zWuoE3q0KT5TUjmPKTb1lw 密碼：pp42

java執行緒池ThreadPoolExecutor類使用詳解

在《阿里巴巴java開發手冊》中指出了執行緒資源必須通過執行緒池提供，不允許在應用中自行顯示的建立執行緒，這樣一方面是執行緒的建立更加規範，可以合理控制開闢執行緒的數量；另一方面執行緒的細節管理交給執行緒

深度解讀 java 執行緒池設計思想及原始碼實現

轉自 https://javadoop.com/2017/09/05/java-thread-pool/hmsr=toutiao.io&utm_medium=toutiao.io&utm_source=toutiao.io

Java併發包原始碼學習系列：掛起與喚醒執行緒LockSupport工具類

技術標籤：Java併發佇列多執行緒java併發程式設計 文章目錄 LockSupport概述park與unpark相關方法中斷演示blocker的作用測試無blocker測試帶blocker

java執行緒間通訊的幾種方式

技術標籤：面試併發程式設計多執行緒併發程式設計面試 併發程式設計中，我們可能會遇到這樣一個場景

java 執行緒是如何在幾種狀態之間轉換的？

執行緒也有自己的生命週期，在 Java 中執行緒的生命週期中一共有 6 種常見狀態。

Android 執行緒池ThreadPoolExecutor類

一、ThreadPoolExecutor類使用 1. 建立執行緒池原始碼： /** * Creates a new {@code ThreadPoolExecutor} with the given initial

工具類--執行緒相關工具類

package com.jyc.common.utils;import java.util.concurrent.CancellationException;import java.util.concurrent.ExecutionException;import java.util.concurrent.ExecutorService;import java.util.concurrent.F