多線程池、飽和策略詳解

時間  2019-11-09
標籤 多線程 飽和 策略 詳解 欄目 Java 简体版
原文   原文鏈接

1、序言

當咱們須要使用線程的時候，咱們能夠新建一個線程，而後顯式調用線程的start()方法，這樣實現起來很是簡便，但在某些場景下存在缺陷：若是須要同時執行多個任務（即併發的線程數量不少），頻繁地建立線程會下降系統的效率，由於建立和銷燬線程均須要必定的時間。html

線程池可使線程獲得複用，所謂線程複用就是線程在執行完一個任務後並不被銷燬，該線程能夠繼續執行其餘的任務。java.lang.concurrent包中的Executors類爲咱們建立線程池提供了方便。java

2、Executors的簡單使用示例

此處咱們先來看一個簡單的例子，以下：面試

package com.soft; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; public class ExecutorsDemo { public static void main(String[] args) throws InterruptedException, ExecutionException { // ExecutorService executor = Executors.newSingleThreadExecutor(); // ExecutorService executor = Executors.newCachedThreadPool();
        ExecutorService executor = Executors.newFixedThreadPool(5);
        Thread.sleep(5*1000);//方便監控工具能捕獲到
        for (int i = 0; i < 10; i++) { final int no = i;
            Runnable runnable = new Runnable() { public void run() { try {
                        System.out.println("into" + no);
                        Thread.sleep(1000L);
                        System.out.println("end" + no);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            };
            executor.execute(runnable);//ExecutorService有一個execute()方法，這個方法的參數是Runnable類型，經過execute(Runnable)方法便可將一個任務添加到線程池，任務的執行方法是Runnable類型對象的run()方法。
        }//End for
 executor.shutdown();
        System.out.println("Thread Main End!");
    }
}

複製代碼

其運行結果以下：算法

into0
into3
Thread Main End! into4
into1
into2
end0
into5
end3
end1
end4
into8
into6
into7
end2
into9
end5
end7
end8
end6
end9

複製代碼

解說：這個例子應該很容易看懂，從運行結果來看，在任意某一時刻只有5個線程在執行，這是由於上述代碼經過Executors.newFixedThreadPool(5)語句建立了一個固定長度的線程池（長度爲5），一個結束以後另再一個纔開始執行。swift

3、Executors提供的線程池

Executors是線程的工廠類，也能夠說是一個線程池工具類，它調用其內部靜態方法（如newFixedThreadPool()等）便可建立一個線程池，經過參數設置，Executors提供不一樣的線程池機制。bash

image

4、簡述線程池的屬性

image

5、詳解ThreadPoolExecutor##

上文提到能夠經過顯式的ThreadPoolExecutor構造函數來構造特定形式的線程池，ThreadPoolExecutor是java.util.concurrent包之內部線程池的形式對外提供線程池管理、線程調度等服務，此處咱們來了解一下ThreadPoolExecutor微信

  • （1）通常使用方式：
ExecutorService exec = new ThreadPoolExecutor(8, 8, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());

複製代碼

下文詳解此示例涉及的一些內容數據結構

  • 2）構造函數的聲明：
public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,  
                              TimeUnit unit,  
                              BlockingQueue<Runnable> workQueue,  
                              ThreadFactory threadFactory,  
                              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;  
    }

複製代碼
  • （3）函數參數說明：
參數名 表明含義
corePoolSize 線程池的基本大小（核心線程池大小）
maximumPoolSize 線程池的最大大小
keepAliveTime 線程池中超過corePoolSize數目的空閒線程的最大存活時間
unit keepAliveTime參數的時間單位
workQueue 任務阻塞隊列
threadFactory 新建線程的工廠
handler 當提交的任務數超過maxmumPoolSize與workQueue之和時，任務會交給RejectedExecutionHandler來處理

進一步解說：併發

  • A、當提交新任務時，若線程池大小小於corePoolSize，將建立一個新的線程來執行任務，即便此時線程池中存在空閒線程；ide

  • B、當提交新任務時，若線程池達到corePoolSize大小，新提交的任務將被放入workQueue中，等待線程池調度執行；

  • C、當提交新任務時，若workQueue已滿，且maximumPoolSize>corePoolSize，將建立新的線程來執行任務；

  • D、當提交新任務時，若任務總數超過maximumPoolSize，新提交的任務將由RejectedExecutionHandler來處理；

  • E、當線程池中的線程數超過corePoolSize時，若線程的空閒時間達到keepAliveTime，則關閉空閒線程

  • （4）任務阻塞隊列選擇機制

image

  • （5）簡述SynchronousQueue

image

注：此處貼出SynchronousQueue的使用示例，示例中使用了Semaphore，更多關於SynchronousQueue及Semaphore的內容請參考其餘文章

package com.test; import java.util.concurrent.Semaphore; import java.util.concurrent.SynchronousQueue; /* * 程序中有10個線程來消費生成者產生的數據，這些消費者都調用TestDo.doSome()方法去進行處理，
 * 每一個消費者都須要一秒才能處理完，程序應保證這些消費者線程依次有序地消費數據，只有上一個消費者消費完後，
 * 下一個消費者才能消費數據，下一個消費者是誰均可以，但要保證這些消費者線程拿到的數據是有順序的。 */
public class SynchronousQueueTest { public static void main(String[] args) {

        System.out.println("begin:" + (System.currentTimeMillis() / 1000)); // 定義一個Synchronous
        final SynchronousQueue<String> sq = new SynchronousQueue<String>(); // 定義一個數量爲1的信號量，其做用至關於一個互斥鎖
        final Semaphore sem = new Semaphore(1); for (int i = 0; i < 10; i++) { new Thread(new Runnable() { public void run() { try {
                        sem.acquire();
                        String input = sq.take();
                        String output = TestDo.doSome(input);//內部類
                        System.out.println(Thread.currentThread().getName()+ ":" + output);
                        sem.release();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }).start();
        } for (int i = 0; i < 10; i++) {
            String input = i + ""; //此處將i變成字符串
            try {
                sq.put(input);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }//End main
 } class TestDo { public static String doSome(String input) { try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        String output = input + ":" + (System.currentTimeMillis() / 1000); return output;
    }
}

複製代碼

上述代碼的運行結果以下：

begin:1458954798 Thread-0:0:1458954799 Thread-1:1:1458954800 Thread-2:2:1458954801 Thread-3:3:1458954802 Thread-4:4:1458954803 Thread-5:5:1458954804 Thread-6:6:1458954805 Thread-7:7:1458954806 Thread-8:8:1458954807 Thread-9:9:1458954808

複製代碼

從上述結果看，上例在任意某一時刻只有一個線程在執行，且只有前一個線程執行完下一個線程纔開始

6、飽和策略（線程池任務拒絕策略）

上文提到ThreadPoolExecutor構造函數的RejectedExecutionHandler handler參數，該參數表示當提交的任務數超過maxmumPoolSize與workQueue之和時，任務會交給RejectedExecutionHandler來處理，此處咱們來具體瞭解一下

*（1）四種飽和策略

image

image

image

*（2）源碼分析：

RejectedExecutionHandler這個接口是用來處理被丟棄的線程的異常處理接口，其源碼以下：

public interface RejectedExecutionHandler{ //被線程池丟棄的線程處理機制
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) ;
}

複製代碼
  • AbortPolicy（停止策略）繼承RejectedExecutionHandler接口，其源碼以下：
public static class AbortPolicy implements RejectedExecutionHandler{ public AbortPolicy(){} //直接拋出異常
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { throw new RejectedExecutionException("Task"+r.toString()+"rejected from"+executor.toString());
    }
}

複製代碼

咱們能夠本身實現RejectedExecutionHandler接口，將實現類做爲線程丟棄處理類，代碼以下：

package com.test; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadPoolExecutor; public class RejectedExecutionHandlerDemo implements RejectedExecutionHandler{

    @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // TODO Auto-generated method stub
        System.out.println("線程信息"+r.toString()+"被遺棄的線程池："+executor.toString());
    }

}

複製代碼

7、定製ThreadPoolExecutor

*（1）經過修改參數的方式達到定製目的

image

image

*（2）經過自定義方式（封裝各類參數）達到定製目的

import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.ThreadFactory; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; public class CustomThreadPoolExecutor { private ThreadPoolExecutor pool = null; /** * 線程池初始化方法 
     *  
     * corePoolSize 核心線程池大小----10 
     * maximumPoolSize 最大線程池大小----30 
     * keepAliveTime 線程池中超過corePoolSize數目的空閒線程最大存活時間----30+單位TimeUnit 
     * TimeUnit keepAliveTime時間單位----TimeUnit.MINUTES 
     * workQueue 阻塞隊列----new ArrayBlockingQueue<Runnable>(10)====10容量的阻塞隊列 
     * threadFactory 新建線程工廠----new CustomThreadFactory()====定製的線程工廠 
     * rejectedExecutionHandler 當提交任務數超過maxmumPoolSize+workQueue之和時, 
     *                          即當提交第41個任務時(前面線程都沒有執行完,此測試方法中用sleep(100)), 
     *                          任務會交給RejectedExecutionHandler來處理 */  
    public void init() {  
        pool = new ThreadPoolExecutor( 10, 30, 30,  
                TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(10), new CustomThreadFactory(),new CustomRejectedExecutionHandler());  
    } public void destory() { if(pool != null) {  
            pool.shutdownNow();  
        }  
    } public ExecutorService getCustomThreadPoolExecutor() { return this.pool;  
    } private class CustomThreadFactory implements ThreadFactory { private AtomicInteger count = new AtomicInteger(0);  

        @Override public Thread newThread(Runnable r) {  
            Thread t = new Thread(r);  
            String threadName = CustomThreadPoolExecutor.class.getSimpleName() + count.addAndGet(1);  
            System.out.println(threadName);  
            t.setName(threadName); return t;  
        }  
    } private class CustomRejectedExecutionHandler implements RejectedExecutionHandler {  

        @Override public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { // 記錄異常 // 報警處理等 
            System.out.println("error.............");  
        }  
    } // 測試構造的線程池 
    public static void main(String[] args) {  
        CustomThreadPoolExecutor exec = new CustomThreadPoolExecutor(); // 1.初始化 
 exec.init();  

        ExecutorService pool = exec.getCustomThreadPoolExecutor(); for(int i=1; i<100; i++) {  
            System.out.println("提交第" + i + "個任務!");  
            pool.execute(new Runnable() {  
                @Override public void run() { try {  
                        Thread.sleep(300);  
                    } catch (InterruptedException e) {  
                        e.printStackTrace();  
                    }  
                    System.out.println("running=====");  
                }  
            });  
        } // 2.銷燬----此處不能銷燬,由於任務沒有提交執行完,若是銷燬線程池,任務也就沒法執行了 // exec.destory(); 

        try {  
            Thread.sleep(10000);  
        } catch (InterruptedException e) {  
            e.printStackTrace();  
        }  
    }  
}

複製代碼

咱們能夠看到上述代碼是經過init()方法對ThreadPoolExecutor構造函數進行了一些自定義設置，getCustomThreadPoolExecutor()方法返回init()方法配置的ThreadPoolExecutor對象實例（即線程池引用）

補充：##

ThreadPoolExecutor構造函數有一個參數ThreadFactory threadFactory，前文提到該參數是新建線程的工廠，此處進一步解說該參數。

ThreadFactory是java.util.concurrent包下建立線程工廠的接口，ThreadFactory接口源碼以下：

public interface ThreadFactory {  
    Thread newThread(Runnable r);  
}

複製代碼

JDK線程池：Executors.newSingleThreadExecutor、``Executors.newFixedThreadPool等由一個ThreadFactory來建立新的線程，默認狀況下爲Executors.defaultThreadFactory()，咱們能夠採用自定義的ThreadFactory工廠，增長對線程建立與銷燬等更多的控制（好比上述代碼中的內部類CustomThreadFactory即爲新建線程的模板）

此處簡單說起一下，讀者欲瞭解更多內容能夠參考如下文章

8、擴展ThreadPoolExecutor##

image

9、源碼視角##

從源碼視角分析Executors、ThreadPoolExecutor、ExecuteService、Executor之間的關係，此處簡單說起，讀者可查看下一節「參考資料」以瞭解相關內容

  • （1）Executors

從Java5開始新增了Executors類，它有幾個靜態工廠方法用來建立線程池，這些靜態工廠方法返回一個ExecutorService類型的值，此值即爲線程池的引用。

  • （2）Executor

Executor是一個接口，裏面只有一個方法

public interface Executor { void execute(Runnable command);
}

複製代碼
  • （3）ExecuteService

ExecuteService也是一個接口，其定義以下：

public interface ExecutorService extends Executor {...}

  • （4）ThreadPoolExecutor繼承AbstractExecutorService，AbstractExecutorService實現ExecutorService接口

public class ThreadPoolExecutor extends AbstractExecutorService {...}

public abstract class AbstractExecutorService implements ExecutorService {...}

10、ExecutorService的生命週期##

在本文最開始的那個示例中，有一句代碼，以下：

executor.shutdown();

該語句並非終止線程的運行，而是禁止在這個executor中添加新的任務，下文描述了該語句對於ExecutorService的意義。

image

11、參考資料##

本文僅簡單闡述了Java併發中關於Executors及ThreadPoolExecutor的內容，此處貼出一些優質文章以供讀者閱覽

*（1）blog.csdn.net/xiamizy/art…

*（2）www.cnblogs.com/dolphin0520…

*（3）www.cnblogs.com/yezhenhan/a…

*（4）www.cnblogs.com/guguli/p/51…

給你們推薦一個iOS技術微信交流羣!羣內提供數據結構與算法、底層進階、swift、逆向、底層面試題整合文檔等免費資料！因爲羣裏已到達200人 可加我微信邀請你們進羣

image.png
相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程