面試官問如何構建動態線程池應對流量洪峯，咱們聊了三十分鐘

時間  2021-08-14
標籤 java node git github sql 數據庫 緩存 微信 markdown 架構 欄目 快樂工作 简体版
原文   原文鏈接

歡迎你們關注公衆號「JAVA前線」查看更多精彩分享文章，主要包括源碼分析、實際應用、架構思惟、職場分享、產品思考等等，同時歡迎你們加我微信「java_front」一塊兒交流學習java


0 文章概述

流量洪峯是互聯網生產環境常常遇到的狀況，例如某個時間點進行商品搶購活動，或者某個時間點集中觸發定時任務，這些場景都有可能引起流量洪峯，因此如何應對流量洪峯咱們必須面對的問題。node

縱向維度咱們能夠從代理層、WEB層、服務層、緩存層、數據層進行思考，橫向維度咱們能夠從高頻檢測、緩存前置、節點冗餘、服務降級等方向進行思考。本文咱們從服務層動態線程池這個角度思考應對流量洪峯。git

動態線程池是指咱們能夠根據流量的不一樣調節線程池某些參數，例如能夠在業務低峯期調低線程數，在業務高峯期調高線程數增長處理線程從而應對流量洪峯。本文咱們結合Apollo和線程池實現一個動態線程池。github

如何實現動態線程池.jpeg


1 線程池基礎

1.1 七個參數

咱們首先回顧一下Java線程池七大參數，這對後續設置線程池參數有幫助。咱們查看ThreadPoolExecutor構造函數以下：sql

public class ThreadPoolExecutor extends AbstractExecutorService {
    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.acc = System.getSecurityManager() == null ?
                   null :
                   AccessController.getContext();
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.workQueue = workQueue;
        this.keepAliveTime = unit.toNanos(keepAliveTime);
        this.threadFactory = threadFactory;
        this.handler = handler;
    }
}
複製代碼

corePoolSize

線程池核心線程數，類比業務大廳開設的固定窗口。例如業務大廳開設2個固定窗口，那麼這兩個窗口不會關閉，全天都會進行業務辦理數據庫

workQueue

存儲已提交但還沒有執行的任務，類比業務大廳等候區。例如業務大廳一開門進來不少顧客，2個固定窗口進行業務辦理，其餘顧客到等候區等待緩存

maximumPoolSize

線程池能夠容納同時執行最大線程數，類比業務大廳最大窗口數。例如業務大廳最大窗口數是5個，業務員看到2個固定窗口和等候區都滿了，能夠臨時增長3個窗口微信

keepAliveTime

非核心線程數存活時間。當業務不忙時剛纔新增的3個窗口須要關閉，空閒時間超過keepAliveTime空閒會被關閉markdown

unit

keepAliveTime存活時間單位架構

threadFactory

線程工廠能夠用來指定線程名

handler

線程池線程數已達到maximumPoolSize且隊列已滿時執行拒絕策略。例如業務大廳5個窗口所有處於忙碌狀態且等候區已滿，業務員根據實際狀況選擇拒絕策略


1.2 四種拒絕策略

(1) AbortPolicy

默認策略直接拋出RejectExecutionException阻止系統正常運行

/** * AbortPolicy * * @author 微信公衆號「JAVA前線」 * */
public class AbortPolicyTest {
    public static void main(String[] args) {
        int coreSize = 1;
        int maxSize = 2;
        int queueSize = 1;
        AbortPolicy abortPolicy = new ThreadPoolExecutor.AbortPolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, maxSize, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(queueSize), Executors.defaultThreadFactory(), abortPolicy);
        for (int i = 0; i < 100; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " -> run");
                }
            });
        }
    }
}
複製代碼

程序執行結果：

pool-1-thread-1 -> run
pool-1-thread-2 -> run
pool-1-thread-1 -> run
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task com.xy.juc.threadpool.reject.AbortPolicyTest$1@70dea4e rejected from java.util.concurrent.ThreadPoolExecutor@5c647e05[Running, pool size = 2, active threads = 0, queued tasks = 0, completed tasks = 2]
	at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
	at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
	at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
	at com.xy.juc.threadpool.reject.AbortPolicyTest.main(AbortPolicyTest.java:21)
複製代碼

(2) CallerRunsPolicy

任務回退給調用者本身運行

/** * CallerRunsPolicy * * @author 微信公衆號「JAVA前線」 * */
public class CallerRunsPolicyTest {
    public static void main(String[] args) {
        int coreSize = 1;
        int maxSize = 2;
        int queueSize = 1;
        CallerRunsPolicy callerRunsPolicy = new ThreadPoolExecutor.CallerRunsPolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, maxSize, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(queueSize), Executors.defaultThreadFactory(), callerRunsPolicy);
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " -> run");
                }
            });
        }
    }
}
複製代碼

程序執行結果：

main -> run
pool-1-thread-1 -> run
pool-1-thread-2 -> run
pool-1-thread-1 -> run
main -> run
main -> run
pool-1-thread-2 -> run
pool-1-thread-1 -> run
main -> run
pool-1-thread-2 -> run
複製代碼

(3) DiscardOldestPolicy

拋棄隊列中等待最久的任務不會拋出異常

/** * DiscardOldestPolicy * * @author 微信公衆號「JAVA前線」 * */
public class DiscardOldestPolicyTest {
    public static void main(String[] args) {
        int coreSize = 1;
        int maxSize = 2;
        int queueSize = 1;
        DiscardOldestPolicy discardOldestPolicy = new ThreadPoolExecutor.DiscardOldestPolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, maxSize, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(queueSize), Executors.defaultThreadFactory(), discardOldestPolicy);
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " -> run");
                }
            });
        }
    }
}
複製代碼

程序執行結果：

pool-1-thread-1 -> run
pool-1-thread-2 -> run
pool-1-thread-1 -> run
複製代碼

(4) DiscardPolicy

直接丟棄任務不會拋出異常

/** * DiscardPolicy * * @author 微信公衆號「JAVA前線」 * */
public class DiscardPolicyTest {
    public static void main(String[] args) {
        int coreSize = 1;
        int maxSize = 2;
        int queueSize = 1;
        DiscardPolicy discardPolicy = new ThreadPoolExecutor.DiscardPolicy();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(coreSize, maxSize, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(queueSize), Executors.defaultThreadFactory(), discardPolicy);
        for (int i = 0; i < 10; i++) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName() + " -> run");
                }
            });
        }
    }
}
複製代碼

程序執行結果：

pool-1-thread-1 -> run
pool-1-thread-2 -> run
pool-1-thread-1 -> run
複製代碼

1.3 修改參數

若是初始化線程池完成後，咱們是否能夠修改線程池某些參數呢？答案是能夠。咱們選擇線程池提供的四個修改方法進行源碼分析。

(1) setCorePoolSize

public class ThreadPoolExecutor extends AbstractExecutorService {
    public void setCorePoolSize(int corePoolSize) {
        if (corePoolSize < 0)
            throw new IllegalArgumentException();
        // 新核心線程數減去原核心線程數
        int delta = corePoolSize - this.corePoolSize;
        // 新核心線程數賦值
        this.corePoolSize = corePoolSize;
        // 若是當前線程數大於新核心線程數
        if (workerCountOf(ctl.get()) > corePoolSize)
            // 中斷空閒線程
            interruptIdleWorkers();
        // 若是須要新增線程則經過addWorker增長工做線程
        else if (delta > 0) {
            int k = Math.min(delta, workQueue.size());
            while (k-- > 0 && addWorker(null, true)) {
                if (workQueue.isEmpty())
                    break;
            }
        }
    }
}
複製代碼

(2) setMaximumPoolSize

public class ThreadPoolExecutor extends AbstractExecutorService {
    public void setMaximumPoolSize(int maximumPoolSize) {
        if (maximumPoolSize <= 0 || maximumPoolSize < corePoolSize)
            throw new IllegalArgumentException();
        this.maximumPoolSize = maximumPoolSize;
		// 若是當前線程數量大於新最大線程數量
        if (workerCountOf(ctl.get()) > maximumPoolSize)
			// 中斷空閒線程
            interruptIdleWorkers();
    }
}
複製代碼

(3) setKeepAliveTime

public class ThreadPoolExecutor extends AbstractExecutorService {
    public void setKeepAliveTime(long time, TimeUnit unit) {
        if (time < 0)
            throw new IllegalArgumentException();
        if (time == 0 && allowsCoreThreadTimeOut())
            throw new IllegalArgumentException("Core threads must have nonzero keep alive times");
        long keepAliveTime = unit.toNanos(time);
        // 新超時時間減去原超時時間
        long delta = keepAliveTime - this.keepAliveTime;
        this.keepAliveTime = keepAliveTime;
        // 若是新超時時間小於原超時時間
        if (delta < 0)
            // 中斷空閒線程
            interruptIdleWorkers();
    }
}
複製代碼

(4) setRejectedExecutionHandler

public class ThreadPoolExecutor extends AbstractExecutorService {
    public void setRejectedExecutionHandler(RejectedExecutionHandler handler) {
        if (handler == null)
            throw new NullPointerException();
        // 設置拒絕策略
        this.handler = handler;
    }
}
複製代碼

如今咱們知道線程池系統上述調整參數的方法，但僅僅分析到此是不夠的，由於若是沒有動態調整參數的方法，那麼每次修改必須從新發布才能夠生效，那麼有沒有方法不用發佈就能夠動態調整線程池參數呢？


2 Apollo配置中心

2.1 核心原理

Apollo是攜程框架部門研發的分佈式配置中心，可以集中化管理應用不一樣環境、不一樣集羣的配置，配置修改後可以實時推送到應用端，而且具有規範的權限、流程治理等特性，適用於微服務配置管理場景。Apollo開源地址以下：

https://github.com/ctripcorp/apollo
複製代碼

咱們在使用配置中心時第一步用戶在配置中心修改配置項，第二步配置中心通知Apollo客戶端有配置更新，第三步Apollo客戶端從配置中心拉取最新配置，更新本地配置並通知到應用，官網基礎模型圖以下：

01 基礎結構.jpg


那麼配置中心配置項發生變化客戶端如何感知呢？分爲推和拉兩種方式。推依賴客戶端和服務端保持了一個長鏈接，發生數據變化時服務端推送信息給客戶端，這就是長輪詢機制。拉依賴客戶端定時從配置中心服務端拉取應用最新配置，這是一個fallback機制。官網客戶端設計圖以下：

02 客戶端設計.jpg


本文重點分析配置更新推送方式，咱們首先看官網服務端設計圖：


03 服務端設計.jpg


ConfigService模塊提供配置的讀取推送等功能，服務對象是Apollo客戶端。AdminService模塊提供配置的修改發佈等功能，服務對象是Portal模塊即管理界面。須要說明Apollo並無引用消息中間件，圖中發送異步消息是指ConfigService定時掃描異步消息數據表：

04 推送方式.jpg

消息數據保存在MySQL消息表：

CREATE TABLE `releasemessage` (
  `Id` int(11) unsigned NOT NULL AUTO_INCREMENT COMMENT '自增主鍵',
  `Message` varchar(1024) NOT NULL DEFAULT '' COMMENT '發佈的消息內容',
  `DataChange_LastTime` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '最後修改時間',
  PRIMARY KEY (`Id`),
  KEY `DataChange_LastTime` (`DataChange_LastTime`),
  KEY `IX_Message` (`Message`(191))
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8mb4 COMMENT='發佈消息'
複製代碼

Apollo核心原理本文暫且談到這裏，後續我會寫文章經過源碼分析Apollo長輪詢機制請繼續關注。


2.2 實例分析

2.2.1 服務端安裝

服務端關鍵步驟是導入數據庫和修改端口號，具體步驟請參看官方網站：

https://ctripcorp.github.io/apollo/#/zh/deployment/quick-start
複製代碼

啓動成功後訪問地址：

http://localhost:8070
複製代碼

05 訪問首頁.jpg


輸入用戶名apollo、密碼admin登陸：


06 界面1.jpg


點擊進入我建立myApp項目，咱們看到在DEV環境、default集羣、application命名空間包含一個timeout配置項，100是這個配置項的值，下面咱們在應用程序讀取這個配置項：


07 界面2.jpg


2.2.2 應用程序

(1) 引入依賴

<dependencies>
    <dependency>
	<groupId>com.ctrip.framework.apollo</groupId>
	<artifactId>apollo-client</artifactId>
	<version>1.7.0</version>
    </dependency>
</dependencies>	
複製代碼

(2) 簡單實例

public class GetApolloConfigTest extends BaseTest {

    /** * -Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080 * * myApp+DEV+default+application */
    @Test
    public void testGet() throws InterruptedException {
        Config appConfig = ConfigService.getAppConfig();
        while (true) {
            String value = appConfig.getProperty("timeout", "200");
            System.out.println("timeout=" + value);
            TimeUnit.SECONDS.sleep(1);
        }
    }
}
複製代碼

由於上述程序是經過while(true)不斷獲取配置項的值，因此程序輸出結果以下：

timeout=100
timeout=100
timeout=100
timeout=100
timeout=100
timeout=100
複製代碼

咱們如今把配置項的值改成200程序輸出結果以下:

timeout=100
timeout=100
timeout=100
timeout=100
timeout=200
timeout=200
timeout=200
複製代碼

(3) 監聽實例

生產環境咱們通常不用while(true)監聽變化，而是經過註冊監聽器方式感知變化信息：

public class GetApolloConfigTest extends BaseTest {

    /** * 監聽命名空間變化 * * -Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080 * * myApp+DEV+default+application */
    @Test
    public void testListen() throws InterruptedException {
        Config config = ConfigService.getConfig("application");
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                System.out.println("發生變化命名空間=" + changeEvent.getNamespace());
                for (String key : changeEvent.changedKeys()) {
                    ConfigChange change = changeEvent.getChange(key);
                    System.out.println(String.format("發生變化key=%s,oldValue=%s,newValue=%s,changeType=%s", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType()));
                }
            }
        });
        Thread.sleep(1000000L);
    }
}
複製代碼

咱們如今把timeout值從200改成300，程序輸出結果：

發生變化命名空間=application
發生變化key=timeout,oldValue=200,newValue=300,changeType=MODIFIED
複製代碼

3 動態線程池

如今咱們把線程池和Apollo結合起來構建動態線程池，具有了上述知識編寫起來並不複雜。首先咱們用默認值構建一個線程池，而後線程池會監聽Apollo關於相關配置項，若是相關配置有變化則刷新相關參數。第一步在Apollo配置中心設置三個線程池參數（本文沒有設置拒絕策略其原理一致）：


08 線程數1.jpg


第二步編寫核心代碼：

/** * 動態線程池工廠 * * @author 微信公衆號「JAVA前線」 * */
@Slf4j
@Component
public class DynamicThreadPoolFactory {
    private static final String NAME_SPACE = "threadpool-config";

    /** 線程執行器 **/
    private volatile ThreadPoolExecutor executor;

    /** 核心線程數 **/
    private Integer CORE_SIZE = 10;

    /** 最大值線程數 **/
    private Integer MAX_SIZE = 20;

    /** 等待隊列長度 **/
    private Integer QUEUE_SIZE = 2000;

    /** 線程存活時間 **/
    private Long KEEP_ALIVE_TIME = 1000L;

    /** 線程名 **/
    private String threadName;

    public DynamicThreadPoolFactory() {
        Config config = ConfigService.getConfig(NAME_SPACE);
        init(config);
        listen(config);
    }

    /** * 初始化 */
    private void init(Config config) {
        if (executor == null) {
            synchronized (DynamicThreadPoolFactory.class) {
                if (executor == null) {
                    String coreSize = config.getProperty(KeysEnum.CORE_SIZE.getNodeKey(), CORE_SIZE.toString());
                    String maxSize = config.getProperty(KeysEnum.MAX_SIZE.getNodeKey(), MAX_SIZE.toString());
                    String keepAliveTIme = config.getProperty(KeysEnum.KEEP_ALIVE_TIME.getNodeKey(), KEEP_ALIVE_TIME.toString());
                    BlockingQueue<Runnable> queueToUse = new LinkedBlockingQueue<Runnable>(QUEUE_SIZE);
                    executor = new ThreadPoolExecutor(Integer.valueOf(coreSize), Integer.valueOf(maxSize), Long.valueOf(keepAliveTIme), TimeUnit.MILLISECONDS, queueToUse, new NamedThreadFactory(threadName, true), new AbortPolicyDoReport(threadName));
                }
            }
        }
    }

    /** * 監聽器 */
    private void listen(Config config) {
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                log.info("命名空間發生變化={}", changeEvent.getNamespace());
                for (String key : changeEvent.changedKeys()) {
                    ConfigChange change = changeEvent.getChange(key);
                    String newValue = change.getNewValue();
                    refreshThreadPool(key, newValue);
                    log.info("發生變化key={},oldValue={},newValue={},changeType={}", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType());
                }
            }
        });
    }

    /** * 刷新線程池 */
    private void refreshThreadPool(String key, String newValue) {
        if (executor == null) {
            return;
        }
        if (KeysEnum.CORE_SIZE.getNodeKey().equals(key)) {
            executor.setCorePoolSize(Integer.valueOf(newValue));
            log.info("修改核心線程數key={},value={}", key, newValue);
        }
        if (KeysEnum.MAX_SIZE.getNodeKey().equals(key)) {
            executor.setMaximumPoolSize(Integer.valueOf(newValue));
            log.info("修改最大線程數key={},value={}", key, newValue);
        }
        if (KeysEnum.KEEP_ALIVE_TIME.getNodeKey().equals(key)) {
            executor.setKeepAliveTime(Integer.valueOf(newValue), TimeUnit.MILLISECONDS);
            log.info("修改活躍時間key={},value={}", key, newValue);
        }
    }

    public ThreadPoolExecutor getExecutor(String threadName) {
        return executor;
    }

    enum KeysEnum {

        CORE_SIZE("coreSize", "核心線程數"),

        MAX_SIZE("maxSize", "最大線程數"),

        KEEP_ALIVE_TIME("keepAliveTime", "線程活躍時間")

        ;

        private String nodeKey;
        private String desc;

        KeysEnum(String nodeKey, String desc) {
            this.nodeKey = nodeKey;
            this.desc = desc;
        }

        public String getNodeKey() {
            return nodeKey;
        }

        public void setNodeKey(String nodeKey) {
            this.nodeKey = nodeKey;
        }

        public String getDesc() {
            return desc;
        }

        public void setDesc(String desc) {
            this.desc = desc;
        }
    }
}

/** * 動態線程池執行器 * * @author 微信公衆號「JAVA前線」 * */
@Component
public class DynamicThreadExecutor {

	@Resource
	private DynamicThreadPoolFactory threadPoolFactory;

	public void execute(String bizName, Runnable job) {
		threadPoolFactory.getExecutor(bizName).execute(job);
	}

	public Future<?> sumbit(String bizName, Runnable job) {
		return threadPoolFactory.getExecutor(bizName).submit(job);
	}
}
複製代碼

第三步運行測試用例並結合VisualVM觀察線程數：

/** * 動態線程池測試 * * @author 微信公衆號「JAVA前線」 * */
public class DynamicThreadExecutorTest extends BaseTest {

    @Resource
    private DynamicThreadExecutor dynamicThreadExecutor;

    /** * -Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080 * * myApp+DEV+default+thread-pool */
    @Test
    public void testExecute() throws InterruptedException {
        while (true) {
            dynamicThreadExecutor.execute("bizName", new Runnable() {
                @Override
                public void run() {
                    System.out.println("bizInfo");
                }
            });
            TimeUnit.SECONDS.sleep(1);
        }
    }
}
複製代碼

visualvm1.jpg


咱們在配置中心修改配置項把核心線程數設置爲50，最大線程數設置爲100：


10.jpg


經過VisualVM能夠觀察到線程數顯著上升：


11.jpg


4 文章總結

本文咱們首先介紹了線程池基礎知識，包括七大參數和四個拒絕策略，隨後咱們介紹了Apollo配置中心的原理和應用，最後咱們將線程池和配置中心相結合，實現了動態調整線程數的效果，但願本文對你們有所幫助。


歡迎你們關注公衆號「JAVA前線」查看更多精彩分享文章，主要包括源碼分析、實際應用、架構思惟、職場分享、產品思考等等，同時歡迎你們加我微信「java_front」一塊兒交流學習

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