1. 認(rèn)識(shí)java線程池

1.1 在什么情況下使用線程池？

• 1.單個(gè)任務(wù)處理的時(shí)間比較短
• 2.需處理的任務(wù)的數(shù)量大

1.2 使用線程池的好處:

• 1.減少在創(chuàng)建和銷毀線程上所花的時(shí)間以及系統(tǒng)資源的開銷
• 2.如不使用線程池，有可能造成系統(tǒng)創(chuàng)建大量線程而導(dǎo)致消耗完系統(tǒng)內(nèi)存

1.3 線程池包括以下四個(gè)基本組成部分：

• 1、線程池管理器（ThreadPool）：用于創(chuàng)建并管理線程池，包括 創(chuàng)建線程池，銷毀線程池，添加新任務(wù)；
• 2、工作線程（PoolWorker）：線程池中線程，在沒有任務(wù)時(shí)處于等待狀態(tài)，可以循環(huán)的執(zhí)行任務(wù)；
• 3、任務(wù)接口（Task）：每個(gè)任務(wù)必須實(shí)現(xiàn)的接口，以供工作線程調(diào)度任務(wù)的執(zhí)行，它主要規(guī)定了任務(wù)的入口，任務(wù)執(zhí)行完后的收尾工作，任務(wù)的執(zhí)行狀態(tài)等；
• 4、任務(wù)隊(duì)列（taskQueue）：用于存放沒有處理的任務(wù)。提供一種緩沖機(jī)制。

1.4 線程池的核心參數(shù)

ThreadPoolExecutor 有四個(gè)構(gòu)造方法，前三個(gè)都是調(diào)用最后一個(gè)(最后一個(gè)參數(shù)最全)

 public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
       Executors.defaultThreadFactory(), defaultHandler);
  }
  public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue,
               ThreadFactory threadFactory) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
       threadFactory, defaultHandler);
  }
  public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue,
               RejectedExecutionHandler handler) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
       Executors.defaultThreadFactory(), handler);
  }
  // 都調(diào)用它
  public ThreadPoolExecutor(// 核心線程數(shù)
  int corePoolSize, 
               // 最大線程數(shù)
               int maximumPoolSize, 
               // 閑置線程存活時(shí)間
               long keepAliveTime, 
               // 時(shí)間單位
               TimeUnit unit, 
               // 線程隊(duì)列
               BlockingQueue<Runnable> workQueue, 
               // 線程工廠 
               ThreadFactory threadFactory,        
               // 隊(duì)列已滿,而且當(dāng)前線程數(shù)已經(jīng)超過(guò)最大線程數(shù)時(shí)的異常處理策略       
               RejectedExecutionHandler handler  ) {
    if (corePoolSize < 0 ||
      maximumPoolSize <= 0 ||
      maximumPoolSize < corePoolSize ||
      keepAliveTime < 0)
      throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
      throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
  }

主要參數(shù)

corePoolSize：核心線程數(shù)

• 核心線程會(huì)一直存活，即使沒有任務(wù)需要執(zhí)行
• 當(dāng)線程數(shù)小于核心線程數(shù)時(shí)，即使有線程空閑，線程池也會(huì)優(yōu)先創(chuàng)建新線程處理
• 設(shè)置allowCoreThreadTimeout=true（默認(rèn)false）時(shí)，核心線程會(huì)超時(shí)關(guān)閉

maxPoolSize：最大線程數(shù)

• 當(dāng)線程數(shù)>=corePoolSize，且任務(wù)隊(duì)列已滿時(shí)。線程池會(huì)創(chuàng)建新線程來(lái)處理任務(wù)
• 當(dāng)線程數(shù)=maxPoolSize，且任務(wù)隊(duì)列已滿時(shí)，線程池會(huì)拒絕處理任務(wù)而拋出異常

keepAliveTime：線程空閑時(shí)間

• 當(dāng)線程空閑時(shí)間達(dá)到keepAliveTime時(shí)，線程會(huì)退出，直到線程數(shù)量=corePoolSize
• 如果allowCoreThreadTimeout=true，則會(huì)直到線程數(shù)量=0

workQueue：一個(gè)阻塞隊(duì)列，用來(lái)存儲(chǔ)等待執(zhí)行的任務(wù)，這個(gè)參數(shù)的選擇也很重要，會(huì)對(duì)線程池的運(yùn)行過(guò)程產(chǎn)生重大影響，一般來(lái)說(shuō)，這里的阻塞隊(duì)列有以下幾種選擇：

• ArrayBlockingQueue;
• LinkedBlockingQueue;
• SynchronousQueue;

關(guān)于阻塞隊(duì)列可以看這篇：java 阻塞隊(duì)列

threadFactory：線程工廠，主要用來(lái)創(chuàng)建線程；

rejectedExecutionHandler：任務(wù)拒絕處理器，兩種情況會(huì)拒絕處理任務(wù)：

• 當(dāng)線程數(shù)已經(jīng)達(dá)到maxPoolSize，切隊(duì)列已滿，會(huì)拒絕新任務(wù)
• 當(dāng)線程池被調(diào)用shutdown()后，會(huì)等待線程池里的任務(wù)執(zhí)行完畢，再shutdown。如果在調(diào)用shutdown()和線程池真正shutdown之間提交任務(wù)，會(huì)拒絕新任務(wù)

當(dāng)拒絕處理任務(wù)時(shí)線程池會(huì)調(diào)用rejectedExecutionHandler來(lái)處理這個(gè)任務(wù)。如果沒有設(shè)置默認(rèn)是AbortPolicy，會(huì)拋出異常。ThreadPoolExecutor類有幾個(gè)內(nèi)部實(shí)現(xiàn)類來(lái)處理這類情況：

• AbortPolicy 丟棄任務(wù)，拋運(yùn)行時(shí)異常
• CallerRunsPolicy 執(zhí)行任務(wù)
• DiscardPolicy 忽視，什么都不會(huì)發(fā)生
• DiscardOldestPolicy 從隊(duì)列中踢出最先進(jìn)入隊(duì)列（最后一個(gè)執(zhí)行）的任務(wù)
• 實(shí)現(xiàn)RejectedExecutionHandler接口，可自定義處理器

1.5 Java線程池 ExecutorService

• Executors.newCachedThreadPool 創(chuàng)建一個(gè)可緩存線程池，如果線程池長(zhǎng)度超過(guò)處理需要，可靈活回收空閑線程，若無(wú)可回收，則新建線程。
• Executors.newFixedThreadPool 創(chuàng)建一個(gè)定長(zhǎng)線程池，可控制線程最大并發(fā)數(shù)，超出的線程會(huì)在隊(duì)列中等待。
• Executors.newScheduledThreadPool 創(chuàng)建一個(gè)定長(zhǎng)線程池，支持定時(shí)及周期性任務(wù)執(zhí)行。
• Executors.newSingleThreadExecutor 創(chuàng)建一個(gè)單線程化的線程池，它只會(huì)用唯一的工作線程來(lái)執(zhí)行任務(wù)，保證所有任務(wù)按照指定順序(FIFO, LIFO, 優(yōu)先級(jí))執(zhí)行。

備注：Executors只是一個(gè)工廠類，它所有的方法返回的都是ThreadPoolExecutor、ScheduledThreadPoolExecutor這兩個(gè)類的實(shí)例。

1.6 ExecutorService有如下幾個(gè)執(zhí)行方法

• executorService.execute(Runnable);這個(gè)方法接收一個(gè)Runnable實(shí)例，并且異步的執(zhí)行
• executorService.submit(Runnable)
• executorService.submit(Callable)
• executorService.invokeAny(…)
• executorService.invokeAll(…)

execute(Runnable)

這個(gè)方法接收一個(gè)Runnable實(shí)例，并且異步的執(zhí)行

executorService.execute(new Runnable() {
public void run() {
  System.out.println("Asynchronous task");
}
});
executorService.shutdown();

submit(Runnable)

submit(Runnable)和execute(Runnable)區(qū)別是前者可以返回一個(gè)Future對(duì)象，通過(guò)返回的Future對(duì)象，我們可以檢查提交的任務(wù)是否執(zhí)行完畢，請(qǐng)看下面執(zhí)行的例子：

Future future = executorService.submit(new Runnable() {
public void run() {
  System.out.println("Asynchronous task");
}
});
future.get(); //returns null if the task has finished correctly.

submit(Callable)

submit(Callable)和submit(Runnable)類似，也會(huì)返回一個(gè)Future對(duì)象，但是除此之外，submit(Callable)接收的是一個(gè)Callable的實(shí)現(xiàn)，Callable接口中的call()方法有一個(gè)返回值，可以返回任務(wù)的執(zhí)行結(jié)果，而Runnable接口中的run()方法是void的，沒有返回值。請(qǐng)看下面實(shí)例：

Future future = executorService.submit(new Callable(){
public Object call() throws Exception {
  System.out.println("Asynchronous Callable");
  return "Callable Result";
}
});
System.out.println("future.get() = " + future.get());

如果任務(wù)執(zhí)行完成，future.get()方法會(huì)返回Callable任務(wù)的執(zhí)行結(jié)果。注意，future.get()方法會(huì)產(chǎn)生阻塞。

invokeAny(…)

invokeAny(…)方法接收的是一個(gè)Callable的集合，執(zhí)行這個(gè)方法不會(huì)返回Future，但是會(huì)返回所有Callable任務(wù)中其中一個(gè)任務(wù)的執(zhí)行結(jié)果。這個(gè)方法也無(wú)法保證返回的是哪個(gè)任務(wù)的執(zhí)行結(jié)果，反正是其中的某一個(gè)。

ExecutorService executorService = Executors.newSingleThreadExecutor();
Set<Callable<String>> callables = new HashSet<Callable<String>>();
callables.add(new Callable<String>() {
public String call() throws Exception {
  return "Task 1";
}
});
callables.add(new Callable<String>() {
public String call() throws Exception {
  return "Task 2";
}
});
callables.add(new Callable<String>() {
  public String call() throws Exception {
  return "Task 3";
}
});
String result = executorService.invokeAny(callables);
System.out.println("result = " + result);
executorService.shutdown();

invokeAll(…)

invokeAll(…)與 invokeAny(…)類似也是接收一個(gè)Callable集合，但是前者執(zhí)行之后會(huì)返回一個(gè)Future的List，其中對(duì)應(yīng)著每個(gè)Callable任務(wù)執(zhí)行后的Future對(duì)象。

List<Future<String>> futures = executorService.invokeAll(callables);
for(Future<String> future : futures){
System.out.println("future.get = " + future.get());
}
executorService.shutdown();

2. 在springBoot中使用java線程池ExecutorService

2.1 springBoot 的使用配置

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
 * 數(shù)據(jù)收集配置，主要作用在于Spring啟動(dòng)時(shí)自動(dòng)加載一個(gè)ExecutorService對(duì)象.
 * @author Bruce
 * @date 2017/2/22
 * update by Cliff at 2027/11/03
 */
@Configuration
public class ThreadPoolConfig {
  @Bean
  public ExecutorService getThreadPool(){
    return Executors.newFixedThreadPool();
  }
}

2.2 使用

在@service 中注入 ExecutorService 然后就可以直接用了。
  @Autowired
  private ExecutorService executorService;
public void test(){
    executorService.execute(new Runnable() {
      public void run() {
        System.out.println("Asynchronous task");
      }
    });
  }

總結(jié)

以上就是這篇文章的全部?jī)?nèi)容了，希望本文的內(nèi)容對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，謝謝大家對(duì)腳本之家的支持。如果你想了解更多相關(guān)內(nèi)容請(qǐng)查看下面相關(guān)鏈接

最新評(píng)論