Java ThreadPool的使用解析

更新時間：2020年10月20日 14:07:32 作者：flydean

這篇文章主要介紹了Java ThreadPool的使用解析，幫助大家更好的理解和使用Java，感興趣的朋友可以了解下

簡介

在java中，除了單個使用Thread之外，我們還會使用到ThreadPool來構(gòu)建線程池，那么在使用線程池的過程中需要注意哪些事情呢？

一起來看看吧。

java自帶的線程池

java提供了一個非常好用的工具類Executors，通過Executors我們可以非常方便的創(chuàng)建出一系列的線程池：

Executors.newCachedThreadPool,根據(jù)需要可以創(chuàng)建新線程的線程池。線程池中曾經(jīng)創(chuàng)建的線程，在完成某個任務(wù)后也許會被用來完成另外一項任務(wù)。

Executors.newFixedThreadPool(int nThreads) ,創(chuàng)建一個可重用固定線程數(shù)的線程池。這個線程池里最多包含nThread個線程。

Executors.newSingleThreadExecutor() ,創(chuàng)建一個使用單個 worker 線程的 Executor。即使任務(wù)再多，也只用1個線程完成任務(wù)。

Executors.newSingleThreadScheduledExecutor() ,創(chuàng)建一個單線程執(zhí)行程序，它可安排在給定延遲后運(yùn)行命令或者定期執(zhí)行。

提交給線程池的線程要是可以被中斷的

ExecutorService線程池提供了兩個很方便的停止線程池中線程的方法，他們是shutdown和shutdownNow。

shutdown不會接受新的任務(wù)，但是會等待現(xiàn)有任務(wù)執(zhí)行完畢。而shutdownNow會嘗試立馬終止現(xiàn)有運(yùn)行的線程。

那么它是怎么實現(xiàn)的呢？我們看一個ThreadPoolExecutor中的一個實現(xiàn)：

  public List<Runnable> shutdownNow() {
    List<Runnable> tasks;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
      checkShutdownAccess();
      advanceRunState(STOP);
      interruptWorkers();
      tasks = drainQueue();
    } finally {
      mainLock.unlock();
    }
    tryTerminate();
    return tasks;
  }

里面有一個interruptWorkers()方法的調(diào)用，實際上就是去中斷當(dāng)前運(yùn)行的線程。

所以我們可以得到一個結(jié)論，提交到ExecutorService中的任務(wù)一定要是可以被中斷的，否則shutdownNow方法將會失效。

先看一個錯誤的使用例子：

  public void wrongSubmit(){
    Runnable runnable= ()->{
      try(SocketChannel sc = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8080))) {
      ByteBuffer buf = ByteBuffer.allocate(1024);
      while(true){
        sc.read(buf);
      }
      } catch (IOException e) {
        e.printStackTrace();
      }
    };
    ExecutorService pool = Executors.newFixedThreadPool(10);
    pool.submit(runnable);
    pool.shutdownNow();
  }

在這個例子中，運(yùn)行的代碼無法處理中斷，所以將會一直運(yùn)行。

下面看下正確的寫法：

  public void correctSubmit(){
    Runnable runnable= ()->{
      try(SocketChannel sc = SocketChannel.open(new InetSocketAddress("127.0.0.1", 8080))) {
        ByteBuffer buf = ByteBuffer.allocate(1024);
        while(!Thread.interrupted()){
          sc.read(buf);
        }
      } catch (IOException e) {
        e.printStackTrace();
      }
    };
    ExecutorService pool = Executors.newFixedThreadPool(10);
    pool.submit(runnable);
    pool.shutdownNow();
  }

我們需要在while循環(huán)中加上中斷的判斷，從而控制程序的執(zhí)行。

正確處理線程池中線程的異常

如果在線程池中的線程發(fā)生了異常，比如RuntimeException，我們怎么才能夠捕捉到呢？如果不能夠?qū)Ξ惓＿M(jìn)行合理的處理，那么將會產(chǎn)生不可預(yù)料的問題。

看下面的例子：

  public void wrongSubmit() throws InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(10);
    Runnable runnable= ()->{
      throw new NullPointerException();
    };
    pool.execute(runnable);
    Thread.sleep(5000);
    System.out.println("finished!");
  }

上面的例子中，我們submit了一個任務(wù)，在任務(wù)中會拋出一個NullPointerException，因為是非checked異常，所以不需要顯式捕獲，在任務(wù)運(yùn)行完畢之后，我們基本上是不能夠得知任務(wù)是否運(yùn)行成功了。

那么，怎么才能夠捕獲這樣的線程池異常呢？這里介紹大家?guī)讉€方法。

第一種方法就是繼承ThreadPoolExecutor，重寫

 protected void afterExecute(Runnable r, Throwable t) { }

和

protected void terminated() { }

這兩個方法。

其中afterExecute會在任務(wù)執(zhí)行完畢之后被調(diào)用，Throwable t中保存的是可能出現(xiàn)的運(yùn)行時異常和Error。我們可以根據(jù)需要進(jìn)行處理。

而terminated是在線程池中所有的任務(wù)都被調(diào)用完畢之后才被調(diào)用的。我們可以在其中做一些資源的清理工作。

第二種方法就是使用UncaughtExceptionHandler。

Thread類中提供了一個setUncaughtExceptionHandler方法，用來處理捕獲的異常，我們可以在創(chuàng)建Thread的時候，為其添加一個UncaughtExceptionHandler就可以了。

但是ExecutorService執(zhí)行的是一個個的Runnable，怎么使用ExecutorService來提交Thread呢？

別怕， Executors在構(gòu)建線程池的時候，還可以讓我們傳入ThreadFactory，從而構(gòu)建自定義的Thread。

  public void useExceptionHandler() throws InterruptedException {
    ThreadFactory factory =
        new ExceptionThreadFactory(new MyExceptionHandler());
    ExecutorService pool =
        Executors.newFixedThreadPool(10, factory);
    Runnable runnable= ()->{
      throw new NullPointerException();
    };
    pool.execute(runnable);
    Thread.sleep(5000);
    System.out.println("finished!");
  }

  public static class ExceptionThreadFactory implements ThreadFactory {
    private static final ThreadFactory defaultFactory =
        Executors.defaultThreadFactory();
    private final Thread.UncaughtExceptionHandler handler;

    public ExceptionThreadFactory(
        Thread.UncaughtExceptionHandler handler)
    {
      this.handler = handler;
    }

    @Override
    public Thread newThread(Runnable run) {
      Thread thread = defaultFactory.newThread(run);
      thread.setUncaughtExceptionHandler(handler);
      return thread;
    }
  }

  public static class MyExceptionHandler implements Thread.UncaughtExceptionHandler {
    @Override
    public void uncaughtException(Thread t, Throwable e) {

    }
  }

上面的例子有點(diǎn)復(fù)雜了，有沒有更簡單點(diǎn)的做法呢？

有的。ExecutorService除了execute來提交任務(wù)之外，還可以使用submit來提交任務(wù)。不同之處是submit會返回一個Future來保存執(zhí)行的結(jié)果。

  public void useFuture() throws InterruptedException {
    ExecutorService pool = Executors.newFixedThreadPool(10);
    Runnable runnable= ()->{
      throw new NullPointerException();
    };
    Future future = pool.submit(runnable);
    try {
      future.get();
    } catch (InterruptedException e) {
      e.printStackTrace();
    } catch (ExecutionException e) {
      e.printStackTrace();
    }
    Thread.sleep(5000);
    System.out.println("finished!");
  }

當(dāng)我們在調(diào)用future.get()來獲取結(jié)果的時候，異常也會被封裝到ExecutionException，我們可以直接獲取到。

線程池中使用ThreadLocal一定要注意清理

我們知道ThreadLocal是Thread中的本地變量，如果我們在線程的運(yùn)行過程中用到了ThreadLocal，那么當(dāng)線程被回收之后再次執(zhí)行其他的任務(wù)的時候就會讀取到之前被設(shè)置的變量，從而產(chǎn)生未知的問題。

正確的使用方法就是在線程每次執(zhí)行完任務(wù)之后，都去調(diào)用一下ThreadLocal的remove操作。

或者在自定義ThreadPoolExecutor中，重寫beforeExecute(Thread t, Runnable r)方法，在其中加入ThreadLocal的remove操作。

本文的代碼：

https://github.com/ddean2009/learn-java-base-9-to-20/tree/master/security

以上就是Java ThreadPool的使用解析的詳細(xì)內(nèi)容，更多關(guān)于Java ThreadPool的資料請關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: