本文實(shí)例講述了JAVA線(xiàn)程池原理。分享給大家供大家參考，具體如下：

線(xiàn)程池的優(yōu)點(diǎn)

1、線(xiàn)程是稀缺資源，使用線(xiàn)程池可以減少創(chuàng)建和銷(xiāo)毀線(xiàn)程的次數(shù)，每個(gè)工作線(xiàn)程都可以重復(fù)使用。

2、可以根據(jù)系統(tǒng)的承受能力，調(diào)整線(xiàn)程池中工作線(xiàn)程的數(shù)量，防止因?yàn)橄倪^(guò)多內(nèi)存導(dǎo)致服務(wù)器崩潰。

線(xiàn)程池的創(chuàng)建

public ThreadPoolExecutor(int corePoolSize,
               int maximumPoolSize,
               long keepAliveTime,
               TimeUnit unit,
               BlockingQueue<Runnable> workQueue,
               RejectedExecutionHandler handler) 

• corePoolSize：線(xiàn)程池核心線(xiàn)程數(shù)量
• maximumPoolSize:線(xiàn)程池最大線(xiàn)程數(shù)量
• keepAliverTime：當(dāng)活躍線(xiàn)程數(shù)大于核心線(xiàn)程數(shù)時(shí)，空閑的多余線(xiàn)程最大存活時(shí)間
• unit：存活時(shí)間的單位
• workQueue：存放任務(wù)的隊(duì)列
• handler：超出線(xiàn)程范圍和隊(duì)列容量的任務(wù)的處理程序

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

提交一個(gè)任務(wù)到線(xiàn)程池中，線(xiàn)程池的處理流程如下：

1、判斷線(xiàn)程池里的核心線(xiàn)程是否都在執(zhí)行任務(wù)，如果不是（核心線(xiàn)程空閑或者還有核心線(xiàn)程沒(méi)有被創(chuàng)建）則創(chuàng)建一個(gè)新的工作線(xiàn)程來(lái)執(zhí)行任務(wù)。如果核心線(xiàn)程都在執(zhí)行任務(wù)，則進(jìn)入下個(gè)流程。

2、線(xiàn)程池判斷工作隊(duì)列是否已滿(mǎn)，如果工作隊(duì)列沒(méi)有滿(mǎn)，則將新提交的任務(wù)存儲(chǔ)在這個(gè)工作隊(duì)列里。如果工作隊(duì)列滿(mǎn)了，則進(jìn)入下個(gè)流程。

3、判斷線(xiàn)程池里的線(xiàn)程是否都處于工作狀態(tài)，如果沒(méi)有，則創(chuàng)建一個(gè)新的工作線(xiàn)程來(lái)執(zhí)行任務(wù)。如果已經(jīng)滿(mǎn)了，則交給飽和策略來(lái)處理這個(gè)任務(wù)。

線(xiàn)程池的源碼解讀

1、ThreadPoolExecutor的execute()方法

public void execute(Runnable command) {
    if (command == null)
      throw new NullPointerException(); //如果線(xiàn)程數(shù)大于等于基本線(xiàn)程數(shù)或者線(xiàn)程創(chuàng)建失敗，將任務(wù)加入隊(duì)列
    if (poolSize >= corePoolSize || !addIfUnderCorePoolSize(command)) {//線(xiàn)程池處于運(yùn)行狀態(tài)并且加入隊(duì)列成功
      if (runState == RUNNING && workQueue.offer(command)) {
        if (runState != RUNNING || poolSize == 0)
          ensureQueuedTaskHandled(command);
      }//線(xiàn)程池不處于運(yùn)行狀態(tài)或者加入隊(duì)列失敗，則創(chuàng)建線(xiàn)程（創(chuàng)建的是非核心線(xiàn)程）
      else if (!addIfUnderMaximumPoolSize(command))//創(chuàng)建線(xiàn)程失敗，則采取阻塞處理的方式
        reject(command); // is shutdown or saturated
    }
}

2、創(chuàng)建線(xiàn)程的方法：addIfUnderCorePoolSize(command)

private boolean addIfUnderCorePoolSize(Runnable firstTask) {
    Thread t = null;
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
      if (poolSize < corePoolSize && runState == RUNNING)
        t = addThread(firstTask);
    } finally {
      mainLock.unlock();
    }
    if (t == null)
      return false;
    t.start();
    return true;
}

我們重點(diǎn)來(lái)看第7行：

private Thread addThread(Runnable firstTask) {
    Worker w = new Worker(firstTask);
    Thread t = threadFactory.newThread(w);
    if (t != null) {
      w.thread = t;
      workers.add(w);
      int nt = ++poolSize;
      if (nt > largestPoolSize)
        largestPoolSize = nt;
    }
    return t;
}

這里將線(xiàn)程封裝成工作線(xiàn)程worker，并放入工作線(xiàn)程組里，worker類(lèi)的方法run方法：

public void run() {
  try {
    Runnable task = firstTask;
    firstTask = null;
    while (task != null || (task = getTask()) != null) {
      runTask(task);
      task = null;
    }
  } finally {
    workerDone(this);
  }
}

worker在執(zhí)行完任務(wù)后，還會(huì)通過(guò)getTask方法循環(huán)獲取工作隊(duì)里里的任務(wù)來(lái)執(zhí)行。

我們通過(guò)一個(gè)程序來(lái)觀(guān)察線(xiàn)程池的工作原理：

1、創(chuàng)建一個(gè)線(xiàn)程

public class ThreadPoolTest implements Runnable
{
  @Override
  public void run()
  {
    try
    {
      Thread.sleep(300);
    }
    catch (InterruptedException e)
    {
      e.printStackTrace();
    }
  }
}

2、線(xiàn)程池循環(huán)運(yùn)行16個(gè)線(xiàn)程：

public static void main(String[] args)
{
    LinkedBlockingQueue<Runnable> queue =
      new LinkedBlockingQueue<Runnable>(5);
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5, 10, 60, TimeUnit.SECONDS, queue);
    for (int i = 0; i < 16 ; i++)
    {
      threadPool.execute(
        new Thread(new ThreadPoolTest(), "Thread".concat(i + "")));
      System.out.println("線(xiàn)程池中活躍的線(xiàn)程數(shù)： " + threadPool.getPoolSize());
      if (queue.size() > 0)
      {
        System.out.println("----------------隊(duì)列中阻塞的線(xiàn)程數(shù)" + queue.size());
      }
    }
    threadPool.shutdown();
}

執(zhí)行結(jié)果：

線(xiàn)程池中活躍的線(xiàn)程數(shù)： 1
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 2
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 4
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)1
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)2
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)4
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 6
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 7
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 8
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 9
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 10
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)5
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task Thread[Thread15,5,main] rejected from java.util.concurrent.ThreadPoolExecutor@232204a1[Running, pool size = 10, active threads = 10, queued tasks = 5, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2047)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:823)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1369)
    at test.ThreadTest.main(ThreadTest.java:17)

從結(jié)果可以觀(guān)察出：

1、創(chuàng)建的線(xiàn)程池具體配置為：核心線(xiàn)程數(shù)量為5個(gè)；全部線(xiàn)程數(shù)量為10個(gè)；工作隊(duì)列的長(zhǎng)度為5。
2、我們通過(guò)queue.size()的方法來(lái)獲取工作隊(duì)列中的任務(wù)數(shù)。
3、運(yùn)行原理：

剛開(kāi)始都是在創(chuàng)建新的線(xiàn)程，達(dá)到核心線(xiàn)程數(shù)量5個(gè)后，新的任務(wù)進(jìn)來(lái)后不再創(chuàng)建新的線(xiàn)程，而是將任務(wù)加入工作隊(duì)列，任務(wù)隊(duì)列到達(dá)上線(xiàn)5個(gè)后，新的任務(wù)又會(huì)創(chuàng)建新的普通線(xiàn)程，直到達(dá)到線(xiàn)程池最大的線(xiàn)程數(shù)量10個(gè)，后面的任務(wù)則根據(jù)配置的飽和策略來(lái)處理。我們這里沒(méi)有具體配置，使用的是默認(rèn)的配置AbortPolicy:直接拋出異常。
當(dāng)然，為了達(dá)到我需要的效果，上述線(xiàn)程處理的任務(wù)都是利用休眠導(dǎo)致線(xiàn)程沒(méi)有釋放！??！

RejectedExecutionHandler：飽和策略

當(dāng)隊(duì)列和線(xiàn)程池都滿(mǎn)了，說(shuō)明線(xiàn)程池處于飽和狀態(tài)，那么必須對(duì)新提交的任務(wù)采用一種特殊的策略來(lái)進(jìn)行處理。這個(gè)策略默認(rèn)配置是AbortPolicy，表示無(wú)法處理新的任務(wù)而拋出異常。JAVA提供了4中策略：

1、AbortPolicy：直接拋出異常
2、CallerRunsPolicy：只用調(diào)用所在的線(xiàn)程運(yùn)行任務(wù)
3、DiscardOldestPolicy：丟棄隊(duì)列里最近的一個(gè)任務(wù)，并執(zhí)行當(dāng)前任務(wù)。
4、DiscardPolicy：不處理，丟棄掉。

我們現(xiàn)在用第四種策略來(lái)處理上面的程序：

public static void main(String[] args)
{
    LinkedBlockingQueue<Runnable> queue =
      new LinkedBlockingQueue<Runnable>(3);
    RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue,handler);
    for (int i = 0; i < 9 ; i++)
    {
      threadPool.execute(
        new Thread(new ThreadPoolTest(), "Thread".concat(i + "")));
      System.out.println("線(xiàn)程池中活躍的線(xiàn)程數(shù)： " + threadPool.getPoolSize());
      if (queue.size() > 0)
      {
        System.out.println("----------------隊(duì)列中阻塞的線(xiàn)程數(shù)" + queue.size());
      }
    }
    threadPool.shutdown();
}

執(zhí)行結(jié)果：

線(xiàn)程池中活躍的線(xiàn)程數(shù)： 1
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 2
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 2
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)1
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 2
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)2
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 2
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 3
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 4
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3
線(xiàn)程池中活躍的線(xiàn)程數(shù)： 5
----------------隊(duì)列中阻塞的線(xiàn)程數(shù)3

這里采用了丟棄策略后，就沒(méi)有再拋出異常，而是直接丟棄。在某些重要的場(chǎng)景下，可以采用記錄日志或者存儲(chǔ)到數(shù)據(jù)庫(kù)中，而不應(yīng)該直接丟棄。

設(shè)置策略有兩種方式：

1、

RejectedExecutionHandler handler = new ThreadPoolExecutor.DiscardPolicy();
ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue,handler);

2、

ThreadPoolExecutor threadPool = new ThreadPoolExecutor(2, 5, 60, TimeUnit.SECONDS, queue);
threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());

更多java相關(guān)內(nèi)容感興趣的讀者可查看本站專(zhuān)題：《Java進(jìn)程與線(xiàn)程操作技巧總結(jié)》、《Java數(shù)據(jù)結(jié)構(gòu)與算法教程》、《Java操作DOM節(jié)點(diǎn)技巧總結(jié)》、《Java文件與目錄操作技巧匯總》和《Java緩存操作技巧匯總》

希望本文所述對(duì)大家java程序設(shè)計(jì)有所幫助。

最新評(píng)論