線程池的實(shí)現(xiàn)原理是什么

這篇文章給大家介紹線程池的實(shí)現(xiàn)原理是什么，內(nèi)容非常詳細(xì)，感興趣的小伙伴們可以參考借鑒，希望對(duì)大家能有所幫助。

為什么要用線程池

線程是不是越多好？

• 線程在Java中是一個(gè)對(duì)象， 更是操作系統(tǒng)的資源， 線程額創(chuàng)建和銷毀都需要時(shí)間，如果 創(chuàng)建時(shí)間+銷毀時(shí)間>執(zhí)行任務(wù)時(shí)間 就很不合算

• Java對(duì)象占用堆內(nèi)存，操作系統(tǒng)線程占用系統(tǒng)內(nèi)存， 根據(jù)JVM規(guī)范，一個(gè)線程默認(rèn)最大棧大小為1M， 這個(gè)?？臻g是要從操作系統(tǒng)內(nèi)存中分配的，線程過(guò)多會(huì)消耗很多的內(nèi)存

• 操作系統(tǒng)頻繁切換線程上下文會(huì)影響性能

線程池的推出就是為了控制線程數(shù)量

線程池原理 - 概念

• 線程池管理器：

  用于創(chuàng)建并管理線程池， 包括創(chuàng)建線程池， 銷毀線程池， 添加新任務(wù)

• 工作線程：

  線程池中的線程， 在沒(méi)有任務(wù)時(shí)處于等待狀態(tài)， 可以循環(huán)的執(zhí)行任務(wù)

• 任務(wù)接口：

  每個(gè)任務(wù)必須實(shí)現(xiàn)的接口，以供工作線程任務(wù)調(diào)度的執(zhí)行， 它主要規(guī)定了任務(wù)的入口，任務(wù)執(zhí)行完后的收尾工作，任務(wù)的執(zhí)行狀態(tài)等

• 任務(wù)隊(duì)列：

  用于存放沒(méi)有處理的任務(wù)，提供一種緩沖

線程池API - 接口定義和實(shí)現(xiàn)類

線程池API - 方法定義

ExecutorService

cheduledExecutorService

線程池API - Executors工具類

可以自己實(shí)例化線程池， 也可以用Executors創(chuàng)建線程池，方法如下：

• newFixedThreadPool(int nThreads) : 創(chuàng)建一個(gè)固定大小，任務(wù)隊(duì)列無(wú)界的的線程池，核心線程數(shù)=最大線程數(shù)

• newCachedThreadPool() : 創(chuàng)建一個(gè)大小無(wú)界的緩沖線程池，它的任務(wù)隊(duì)列是一個(gè)同步隊(duì)列，任務(wù)加入加入到池中，如果池中有空閑線程， 則用空閑線程執(zhí)行，如無(wú)則創(chuàng)建新線程執(zhí)行，池中的空閑線程超過(guò)60秒， 將被銷毀釋放， 線程數(shù)隨任務(wù)的多少變化，適用于耗時(shí)較小的異步任務(wù)， 池的核心線程數(shù)=0， 最大線程數(shù)=Integer.MAX_VALUE

• newSingleThreadExecutor() : 只有一個(gè)線程來(lái)執(zhí)行無(wú)界任務(wù)隊(duì)列的單一線程池，該線程池確保任務(wù)按加入的順序一個(gè)一個(gè)的依次執(zhí)行，當(dāng)唯一的線程因任務(wù)異常中止時(shí)，將創(chuàng)建一個(gè)新的線程來(lái)繼續(xù)執(zhí)行后續(xù)的任務(wù)， 與newFixedThreadPool(1)的區(qū)別在于，單一線程池的池大小在newSingleThreadExecutor方法中硬編碼，不能再改變

• newScheduledThreadPool(int corePoolSize) : 能定時(shí)執(zhí)行任務(wù)的線程池，該池的核心線程數(shù)由參數(shù)指定，最大線程數(shù)=Integer.MAX_VALUE

線程池原理 - 任務(wù)執(zhí)行過(guò)程

1. 是否達(dá)到核心線程數(shù)量？

   沒(méi)達(dá)到，創(chuàng)建一個(gè)工作線程來(lái)執(zhí)行任務(wù)

2. 工作隊(duì)列是否已滿？

   沒(méi)滿，則將新提交的任務(wù)存儲(chǔ)在任務(wù)隊(duì)列中

3. 是否達(dá)到線程池最大數(shù)量？

   則創(chuàng)建一個(gè)新的線程來(lái)執(zhí)行任務(wù)

4. 最后，執(zhí)行拒絕策略來(lái)處理這個(gè)任務(wù)

代碼示例：

package com.neteasy.demo;

import java.util.List;
import java.util.concurrent.*;

/** 線程池的使用 */
public class Demo7 {

  /**
   * 測(cè)試：提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù),看線程池的狀況
   *
   * @param threadPoolExecutor
   * @throws Exception
   */
  private void testCommon(ThreadPoolExecutor threadPoolExecutor) throws Exception {
    // 測(cè)試：提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù)，看超過(guò)大小的2個(gè)，對(duì)應(yīng)的處理情況
    for (int i = 0; i < 15; i++) {
      int n = i;
      threadPoolExecutor.submit(
          new Runnable() {
            @Override
            public void run() {
              try {
                System.out.println("開(kāi)始執(zhí)行：" + n);
                Thread.sleep(3000L);
                System.out.println("執(zhí)行結(jié)束：" + n);
              } catch (InterruptedException e) {
                e.printStackTrace();
              }
            }
          });
      System.out.println("任務(wù)提交成功：" + i);
    }
    // 查看線程數(shù)量，查看隊(duì)列等待數(shù)量
    Thread.sleep(500L);
    System.out.println("當(dāng)前線程池線程數(shù)量為：" + threadPoolExecutor.getPoolSize());
    System.out.println("當(dāng)前線程池等待的數(shù)量為" + threadPoolExecutor.getQueue().size());
    // 等待15秒，查看線程數(shù)量和隊(duì)列數(shù)量（理論上，會(huì)被超出核心線程數(shù)量的線程自動(dòng)銷毀）
    Thread.sleep(15000L);
    System.out.println("當(dāng)前線程池線程數(shù)量為：" + threadPoolExecutor.getPoolSize());
    System.out.println("當(dāng)前線程池等待的數(shù)量為：" + threadPoolExecutor.getQueue().size());
  }

  /**
   * 1、線程池信息：核心線程數(shù)量5，最大數(shù)量10，無(wú)界隊(duì)列，超出核心線程數(shù)量的線程存活時(shí)間：5秒， 指定拒絕策略的
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest1() throws Exception {
    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(5, 10, 5, TimeUnit.SECONDS, new LinkedBlockingDeque<Runnable>());
    testCommon(threadPoolExecutor);
    // 預(yù)計(jì)結(jié)果：線程池線程數(shù)量為：5,超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
  }

  /**
   * 2、 線程池信息：核心線程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線程數(shù)量的線程存活時(shí)間：5秒， 指定拒絕策略的
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest2() throws Exception {
    // 創(chuàng)建一個(gè) 核心線程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3 的線程池，也就是最大容納13個(gè)任務(wù)。
    // 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(
            5,
            10,
            5,
            TimeUnit.SECONDS,
            new LinkedBlockingDeque<Runnable>(3),
            new RejectedExecutionHandler() {
              @Override
              public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("有任務(wù)被拒絕執(zhí)行了");
              }
            });
    testCommon(threadPoolExecutor);
    // 預(yù)計(jì)結(jié)果：
    // 1、 5個(gè)任務(wù)直接分配線程開(kāi)始執(zhí)行
    // 2、 3個(gè)任務(wù)進(jìn)入等待隊(duì)列
    // 3、 隊(duì)列不夠用，臨時(shí)加開(kāi)5個(gè)線程來(lái)執(zhí)行任務(wù)(5秒沒(méi)活干就銷毀)
    // 4、 隊(duì)列和線程池都滿了，剩下2個(gè)任務(wù)，沒(méi)資源了，被拒絕執(zhí)行。
    // 5、 任務(wù)執(zhí)行，5秒后，如果無(wú)任務(wù)可執(zhí)行，銷毀臨時(shí)創(chuàng)建的5個(gè)線程
  }

  /**
   * 3、 線程池信息：核心線程數(shù)量5，最大數(shù)量5，無(wú)界隊(duì)列，超出核心線程數(shù)量的線程存活時(shí)間：5秒
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest3() throws Exception {
    // 和Executors.newFixedThreadPool(int nThreads)一樣的
    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(
            5, 5, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
    testCommon(threadPoolExecutor);
    // 預(yù)計(jì)結(jié)：線程池線程數(shù)量為：5，超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
  }

  /**
   * 4、 線程池信息：核心線程數(shù)量0，最大數(shù)量Integer.MAX_VALUE，SynchronousQueue隊(duì)列，超出核心線程數(shù)量的線程存活時(shí)間：60秒
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest4() throws Exception {

    /**
     * SynchronousQueue，實(shí)際上它不是一個(gè)真正的隊(duì)列，因?yàn)樗粫?huì)為隊(duì)列中元素維護(hù)存儲(chǔ)空間。與其他隊(duì)列不同的是，它維護(hù)一組線程，這些線程在等待著把元素加入或移出隊(duì)列。
     * 在使用SynchronousQueue作為工作隊(duì)列的前提下，客戶端代碼向線程池提交任務(wù)時(shí)， 而線程池中又沒(méi)有空閑的線程能夠從SynchronousQueue隊(duì)列實(shí)例中取一個(gè)任務(wù)，
     * 那么相應(yīng)的offer方法調(diào)用就會(huì)失敗（即任務(wù)沒(méi)有被存入工作隊(duì)列）。此時(shí)，ThreadPoolExecutor會(huì)新建一個(gè)新的工作者線程用于對(duì)這個(gè)入隊(duì)列失敗的任務(wù)進(jìn)行處理
     * （假設(shè)此時(shí)線程池的大小還未達(dá)到其最大線程池大小maximumPoolSize）。
     */
    // 和Executors.newCachedThreadPool()一樣的
    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(
            0, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>());
    testCommon(threadPoolExecutor);
    // 預(yù)計(jì)結(jié)果：
    // 1、 線程池線程數(shù)量為：15，超出數(shù)量的任務(wù)，其他的進(jìn)入隊(duì)列中等待被執(zhí)行
    // 2、 所有任務(wù)執(zhí)行結(jié)束，60秒后，如果無(wú)任務(wù)可執(zhí)行，所有線程全部被銷毀，池的大小恢復(fù)為0
    Thread.sleep(60000L);
    System.out.println("60秒后，再看線程池中的數(shù)量：" + threadPoolExecutor.getPoolSize());
  }

  /**
   * 5、 定時(shí)執(zhí)行線程池信息：3秒后執(zhí)行，一次性任務(wù)，到點(diǎn)就執(zhí)行
   * 核心線程數(shù)量5，最大數(shù)量Integer.MAX_VALUE，DelayedWorkQueue延時(shí)隊(duì)列，超出核心線程數(shù)量的線程存活時(shí)間：0秒
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest5() throws Exception {
    // 和Executors.newScheduledThreadPool()一樣的
    ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
    threadPoolExecutor.schedule(
        new Runnable() {
          @Override
          public void run() {
            System.out.println("任務(wù)被執(zhí)行，現(xiàn)在時(shí)間：" + System.currentTimeMillis());
          }
        },
        3000,
        TimeUnit.MILLISECONDS);
    System.out.println(
        "定時(shí)任務(wù)，提交成功，時(shí)間是："
            + System.currentTimeMillis()
            + ", 當(dāng)前線程池中線程數(shù)量："
            + threadPoolExecutor.getPoolSize());
    // 預(yù)計(jì)結(jié)果：任務(wù)在3秒后被執(zhí)行一次
  }

  /**
   * 6、 定時(shí)執(zhí)行線程池信息：線程固定數(shù)量5 ，<br>
   * 核心線程數(shù)量5，最大數(shù)量Integer.MAX_VALUE，DelayedWorkQueue延時(shí)隊(duì)列，超出核心線程數(shù)量的線程存活時(shí)間：0秒
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest6() throws Exception {
    ScheduledThreadPoolExecutor threadPoolExecutor = new ScheduledThreadPoolExecutor(5);
    /**
     * 周期性執(zhí)行某一個(gè)任務(wù)，線程池提供了兩種調(diào)度方式，這里單獨(dú)演示一下。測(cè)試場(chǎng)景一樣。測(cè)試場(chǎng)景：提交的任務(wù)需要3秒才能執(zhí)行完畢?？磧煞N不同調(diào)度方式的區(qū)別 效果1：
     * 提交后，2秒后開(kāi)始第一次執(zhí)行，之后每間隔1秒，固定執(zhí)行一次(如果發(fā)現(xiàn)上次執(zhí)行還未完畢，則等待完畢，完畢后立刻執(zhí)行)。
     * 也就是說(shuō)這個(gè)代碼中是，3秒鐘執(zhí)行一次（計(jì)算方式：每次執(zhí)行三秒，間隔時(shí)間1秒，執(zhí)行結(jié)束后馬上開(kāi)始下一次執(zhí)行，無(wú)需等待）
     */
    threadPoolExecutor.scheduleAtFixedRate(
        new Runnable() {
          @Override
          public void run() {
            try {
              Thread.sleep(3000L);
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
            System.out.println("任務(wù)-1 被執(zhí)行，現(xiàn)在時(shí)間：" + System.currentTimeMillis());
          }
        },
        2000,
        1000,
        TimeUnit.MILLISECONDS);

    /**
     * 效果2：提交后，2秒后開(kāi)始第一次執(zhí)行，之后每間隔1秒，固定執(zhí)行一次(如果發(fā)現(xiàn)上次執(zhí)行還未完畢，則等待完畢，等上一次執(zhí)行完畢后再開(kāi)始計(jì)時(shí)，等待1秒)。
     * 也就是說(shuō)這個(gè)代碼鐘的效果看到的是：4秒執(zhí)行一次。（計(jì)算方式：每次執(zhí)行3秒，間隔時(shí)間1秒，執(zhí)行完以后再等待1秒，所以是 3+1）
     */
    threadPoolExecutor.scheduleWithFixedDelay(
        new Runnable() {
          @Override
          public void run() {
            try {
              Thread.sleep(3000L);
            } catch (InterruptedException e) {
              e.printStackTrace();
            }
            System.out.println("任務(wù)-2 被執(zhí)行，現(xiàn)在時(shí)間：" + System.currentTimeMillis());
          }
        },
        2000,
        1000,
        TimeUnit.MILLISECONDS);
  }

  private ThreadPoolExecutor testShutdownCommon() {
    // 創(chuàng)建一個(gè) 核心線程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3的線程池，也就是最大容納13個(gè)任務(wù)。
    // 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
    ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(
            5,
            10,
            5,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(3),
            new RejectedExecutionHandler() {
              @Override
              public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                System.err.println("有任務(wù)被拒絕執(zhí)行了");
              }
            });
    // 測(cè)試：提交15個(gè)執(zhí)行時(shí)間需要3秒的任務(wù)，看超過(guò)大小的2個(gè)，對(duì)應(yīng)的處理情況
    for (int i = 0; i < 15; i++) {
      int n = i;
      threadPoolExecutor.submit(
          new Runnable() {
            @Override
            public void run() {
              try {
                System.out.println("開(kāi)始執(zhí)行：" + n);
                Thread.sleep(3000L);
                System.err.println("執(zhí)行結(jié)束:" + n);
              } catch (InterruptedException e) {
                System.out.println("異常：" + e.getMessage());
              }
            }
          });
      System.out.println("任務(wù)提交成功 :" + i);
    }
    return threadPoolExecutor;
  }

  /**
   * 7、 終止線程：線程池信息：核心線程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線程數(shù)量的線程存活時(shí)間：5秒， 指定拒絕策略的
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest7() throws Exception {
    ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
    // 1秒后終止線程池
    Thread.sleep(1000L);
    threadPoolExecutor.shutdown();
    // 再次提交提示失敗
    threadPoolExecutor.submit(
        new Runnable() {
          @Override
          public void run() {
            System.out.println("追加一個(gè)任務(wù)");
          }
        });
    // 結(jié)果分析
    // 1、 10個(gè)任務(wù)被執(zhí)行，3個(gè)任務(wù)進(jìn)入隊(duì)列等待，2個(gè)任務(wù)被拒絕執(zhí)行
    // 2、調(diào)用shutdown后，不接收新的任務(wù)，等待13任務(wù)執(zhí)行結(jié)束
    // 3、 追加的任務(wù)在線程池關(guān)閉后，無(wú)法再提交，會(huì)被拒絕執(zhí)行
  }

  /**
   * 8、 立刻終止線程：線程池信息：核心線程數(shù)量5，最大數(shù)量10，隊(duì)列大小3，超出核心線程數(shù)量的線程存活時(shí)間：5秒， 指定拒絕策略的
   *
   * @throws Exception
   */
  private void threadPoolExecutorTest8() throws Exception {
    // 創(chuàng)建一個(gè) 核心線程數(shù)量為5，最大數(shù)量為10,等待隊(duì)列最大是3 的線程池，也就是最大容納13個(gè)任務(wù)。
    // 默認(rèn)的策略是拋出RejectedExecutionException異常，java.util.concurrent.ThreadPoolExecutor.AbortPolicy
    ThreadPoolExecutor threadPoolExecutor = testShutdownCommon();
    // 1秒后終止線程池
    Thread.sleep(1000L);
    List<Runnable> shutdownNow = threadPoolExecutor.shutdownNow();
    // 再次提交提示失敗
    threadPoolExecutor.submit(
        new Runnable() {
          @Override
          public void run() {
            System.out.println("追加一個(gè)任務(wù)");
          }
        });
    System.out.println("未結(jié)束的任務(wù)有：" + shutdownNow.size());

    // 結(jié)果分析
    // 1、 10個(gè)任務(wù)被執(zhí)行，3個(gè)任務(wù)進(jìn)入隊(duì)列等待，2個(gè)任務(wù)被拒絕執(zhí)行
    // 2、調(diào)用shutdownnow后，隊(duì)列中的3個(gè)線程不再執(zhí)行，10個(gè)線程被終止
    // 3、 追加的任務(wù)在線程池關(guān)閉后，無(wú)法再提交，會(huì)被拒絕執(zhí)行
  }

  public static void main(String[] args) throws Exception {
    Demo7 demo7 = new Demo7();
    //    demo7.threadPoolExecutorTest1();
    //    demo7.threadPoolExecutorTest2();
    //    demo7.threadPoolExecutorTest3();
    //    demo7.threadPoolExecutorTest4();
    //    demo7.threadPoolExecutorTest5();
    //    demo7.threadPoolExecutorTest6();
    //    demo7.threadPoolExecutorTest7();
    demo7.threadPoolExecutorTest8();
  }
}

關(guān)于線程池的實(shí)現(xiàn)原理是什么就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，可以學(xué)到更多知識(shí)。如果覺(jué)得文章不錯(cuò)，可以把它分享出去讓更多的人看到。