Java併發編程（7）- 線程調度 - 線程池

時間  2019-11-11
標籤 java 併發 編程 線程 調度 欄目 Java 简体版
原文   原文鏈接

線程池

平時有接觸過多線程開發的小夥伴們應該都或多或少都有了解、使用過線程池，而《阿里巴巴 Java 手冊》裏也有一條規範：
Java併發編程（7）- 線程調度 - 線程池java

因而可知線程池的重要性，線程池對於限制應用程序中同一時刻運行的線程數頗有用。由於每啓動一個新線程都會有相應的性能開銷，每一個線程都須要給棧分配一些內存等等。spring

咱們能夠把併發執行的任務傳遞給一個線程池，來替代爲每一個併發執行的任務都啓動一個新的線程。只要池裏有空閒的線程，任務就會分配給一個線程執行。在線程池的內部，任務被插入一個阻塞隊列（Blocking Queue ），線程池裏的線程會去取這個隊列裏的任務。當一個新任務插入隊列時，一個空閒線程就會成功的從隊列中取出任務而且執行它。編程

線程池常常應用在多線程服務器上。每一個經過網絡到達服務器的鏈接都被包裝成一個任務而且傳遞給線程池。線程池的線程會併發的處理鏈接上的請求。緩存

簡單來講使用線程池有如下幾個目的：服務器

  • 線程是稀缺資源，不能頻繁的建立。應當將其放入一個池子中，能夠給其餘任務進行復用，減小對象建立、消亡的開銷，性能好
  • 解耦做用；線程的建立於執行徹底分開，方便維護。
  • 線程池可有效控制最大併發線程數，提升系統資源利用率，同時能夠避免過多資源競爭，避免阻塞
  • 線程池可提供定時執行、按期執行、單線程以及併發數控制等功能

直接new Thread的弊端：網絡

  • 每次new Thread 新建對象，性能差
  • 線程缺少統一管理，可能無限制的新建線程，相互競爭，經常會出現佔用過多的系統資源致使死機或者發生OOM（out of memory 內存溢出），這種問題的緣由不是由於單純的new一個Thread，而是可能由於程序的bug或者設計上的缺陷致使不斷new Thread形成的。
  • 缺乏更多功能，如更多執行、按期執行、線程中斷等

線程池原理：多線程

  • 談到線程池就會想到池化技術，其中最核心的思想就是把寶貴的資源放到一個池子中；每次使用都從裏面獲取，用完以後又放回池子供其餘人使用，有點吃大鍋飯的意思。

線程池類圖：
Java併發編程（7）- 線程調度 - 線程池併發

在上邊的類圖中，最上層就是Executor框架，它是一個根據一組執行策略的調用調度執行和控制異步任務的框架，目的是提供一種將任務提交與任務如何運行分離開的機制。它包含了三個executor接口：框架

  • Executor：運行新任務的簡單接口
  • ExecutorService：擴展了Executor，添加了用來管理執行器生命週期和任務生命週期的方法
  • ScheduledExecutorService：擴展了ExecutorService，支持Future和按期執行任務

在類圖中，咱們最常使用的是ThreadPoolExecutor和Executors，這兩個類均可以建立線程池，其中ThreadPoolExecutor是可定製化的去建立線程池，而Executors則屬因而工具類，該類中已經封裝好了一些建立線程池的方法，直接調用相應的方法便可建立線程。異步

但《阿里巴巴 Java 手冊》裏有一條規範指明不容許使用Executors建立線程池，具體以下：
Java併發編程（7）- 線程調度 - 線程池

能夠說線程池體系裏最爲核心的類是ThreadPoolExecutor，也是功能最強的，ThreadPoolExecutor共有四個構造函數，以下：
Java併發編程（7）- 線程調度 - 線程池

線程池參數

其中最多可傳入七個參數，這七個參數配合起來，構成了線程池強大的功能。參數說明：

corePoolSize：核心線程數量

maximumPoolSize：線程最大線程數

workQueue：阻塞隊列，存儲等待執行的任務，很重要，會對線程池運行過程產生重大影響

keepAliveTime：線程沒有任務執行時最多保持多久時間終止（當線程中的線程數量大於corePoolSize的時候，若是這時沒有新的任務提交核心線程外的線程不會當即銷燬，而是等待，直到等待的時間超過keepAliveTime）

unit：keepAliveTime的時間單位

threadFactory：線程工廠，用來建立線程，若不設置則使用默認的工廠來建立線程，這樣新建立出來的線程會具備相同的優先級，而且是非守護的線程，同時也會設置好名稱

rejectHandler：當拒絕處理任務時(阻塞隊列滿)的策略（AbortPolicy默認策略直接拋出異常、CallerRunsPolicy用調用者所在的線程執行任務、DiscardOldestPolicy丟棄隊列中最靠前的任務並執行當前任務、DiscardPolicy直接丟棄當前任務）

拒絕策略的實現類都在TreadPoolExecutor中：
Java併發編程（7）- 線程調度 - 線程池

咱們來講一下其中corePoolSize、maximumPoolSize、workQueue 這三個參數的關係：

若是運行的線程數量小於corePoolSize的時候，直接建立新線程來處理任務。即便線程池中的其餘線程是空閒的。若是線程池中的線程數量大於corePoolSize且小於maximumPoolSize時，那麼只有當workQueue滿的時候才建立新的線程去處理任務。若是corePoolSize與maximumPoolSize是相同的，那麼建立的線程池大小是固定的。這時若是有新任務提交，且workQueue未滿時，就把請求放入workQueue中，等待空閒線程從workQueue取出任務進行處理。若是須要運行的線程數量大於maximumPoolSize時，而且此時workQueue也滿了，那麼就使用rejectHandler參數所指定的拒絕策略去進行處理。

而後咱們來具體介紹一下 workQueue， 它是保存待執行任務的一個阻塞隊列，當咱們提交一個新的任務到線程池後，線程池會根據當前池中正在運行的線程數量來決定該任務的處理方式。處理方式總共有三種：

一、直接切換（SynchronusQueue）

二、×××隊列（LinkedBlockingQueue），若使用該隊列，那麼線程池中可以建立的最大線程數爲corePoolSize，這時maximumPoolSize就不會起做用了。當線程池中全部的核心線程都是運行狀態的時候，新的任務提交就會放入等待隊列中。

三、有界隊列（ArrayBlockingQueue），使用該隊列能夠將線程池中的最大線程數量限制爲maximumPoolSize參數所指定的值，這種方式可以下降資源消耗，可是這種方式使得線程池對線程調度變的更困難。由於此時線程池與隊列容量都是有限的了，因此想讓線程池處理任務的吞吐率達到一個合理的範圍，又想使咱們的線程調度相對簡單，而且還儘量下降線程池對資源的消耗，那麼咱們就須要合理的設置corePoolSize和maximumPoolSize這兩個參數的值

分配技巧： 若是想下降資源的消耗包括下降cpu使用率、操做系統資源的消耗、上下文切換的開銷等等，能夠設置一個較大的隊列容量和較小的線程池容量，這樣會下降線程池處理任務的吞吐量。若是咱們提交的任務常常發生阻塞，咱們能夠考慮調用相關方法調整maximumPoolSize參數的值。若是咱們的隊列容量較小，一般須要把線程池的容量設置得大一些，這樣cpu的使用率相對來講會高一些。可是若是線程池的容量設置的過大，提升任務的數量過多的時候，併發量會增長，那麼線程之間的調度就是一個須要考慮的問題，這樣反而可能會下降處理任務的吞吐量。

線程池狀態

線程池有五種狀態，線程池狀態轉換過程圖以下：
Java併發編程（7）- 線程調度 - 線程池

  • running：運行狀態，能接受新提交的任務，也能處理阻塞隊列中的任務
  • shutdown：關閉狀態，不能處理新的任務，但卻能夠繼續處理阻塞隊列中已保存的任務。在線程池處於 RUNNING 狀態時，調用 shutdown()方法會使線程池進入到該狀態。（finalize() 方法在執行過程當中也會調用shutdown()方法進入該狀態）；
  • stop：中止狀態，不能接受新任務，也不處理隊列中的任務，會中斷正在處理任務的線程。在線程池處於 RUNNING 或 SHUTDOWN 狀態時，調用 shutdownNow() 方法會使線程池進入到該狀態；
  • tidying：若是全部的任務都已終止了，workerCount (有效線程數) 爲0，線程池進入該狀態後會調用 terminated() 方法進入TERMINATED 狀態。
  • terminated：最終狀態，在terminated() 方法執行完後進入該狀態，默認terminated()方法中什麼也沒有作。

線程池經常使用方法：

方法名 描述
execute() 提交任務，交給線程池執行
submit() 提交任務，可以返回執行結果 execute+Future
shutdown() 關閉線程池，等待任務都執行完
shutdownNow() 馬上關閉線程池，不等待任務執行完
getTaskCount() 線程池已執行和未執行的任務總數
getCompleteTaskCount() 已完成的任務數量
getPoolSize() 線程池當前的線程數量
getActiveCount() 當前線程池中正在執行任務的線程數量

使用Executors建立線程池

上文中咱們提到了可使用Executors工具類方便的建立線程，該類中提供了四種建立線程池的方法，以下：

方法名 描述
newCachedThreadPool 建立一個可緩存線程池，若是線程池長度超過處理須要，可靈活回收空閒線程，若無可回收，則新建線程
newFixedThreadPool 建立一個定長線程池，可控制線程最大併發數，超出的線程會在隊列中等待
newScheduledThreadPool 建立一個定長線程池，支持定時及週期性任務執行
newSingleThreadExecutor 建立一個單線程化的線程池，它只會用惟一的工做線程來執行任務，保證全部任務按照指定順序(FIFO, LIFO, 優先級)執行

newCachedThreadPool使用示例：

@Slf4j
public class ThreadPoolExample1 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        // 若需使用ThreadPoolExecutor裏的方法，則須要進行強轉
//        ThreadPoolExecutor executorService = (ThreadPoolExecutor) Executors.newCachedThreadPool();

        for (int i = 0; i < 10; i++) {
            final int index = i;
            executorService.execute(() -> log.info("task: {}", index));
        }
        executorService.shutdown();
    }
}

newFixedThreadPool使用示例：

@Slf4j
public class ThreadPoolExample2 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);

        for (int i = 0; i < 10; i++) {
            final int index = i;
            executorService.execute(() -> log.info("task: {}", index));
        }
        executorService.shutdown();
    }
}

newSingleThreadExecutor使用示例：

@Slf4j
public class ThreadPoolExample3 {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();

        for (int i = 0; i < 10; i++) {
            final int index = i;
            executorService.execute(() -> log.info("task: {}", index));
        }
        executorService.shutdown();
    }
}

newScheduledThreadPool使用示例：

@Slf4j
public class ThreadPoolExample4 {
    public static void main(String[] args) {
        ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);

        // 延遲3秒執行
        executorService.schedule(() -> log.info("Scheduled run"), 3, TimeUnit.SECONDS);

        // 以指定的速率執行任務，這裏是每隔3秒執行一次任務
        executorService.scheduleAtFixedRate(() -> log.info("Scheduled run"), 1, 3, TimeUnit.SECONDS);

        // 以指定的延遲執行任務，這裏是延遲3秒執行一次任務，使用起來和scheduleAtFixedRate基本同樣
        executorService.scheduleWithFixedDelay(() -> log.info("Scheduled run"), 1, 3, TimeUnit.SECONDS);

        executorService.shutdown();
    }
}

關於延遲執行任務的操做，在Java中還可使用Timer類進行實現，以下：

@Slf4j
public class ThreadPoolExample4 {
    public static void main(String[] args) {
        Timer timer = new Timer();
        // 每隔3秒執行一次任務
        timer.schedule(new TimerTask() {
            @Override
            public void run() {
                log.info("timer task run");
            }
        }, new Date(), 3000);
    }
}

雖然可行，可是並不建議這麼使用，在多線程並行處理定時任務時，Timer運行多個TimeTask的話，只要其中之一沒有捕獲拋出的異常，其它任務便會自動終止運行，使用ScheduledExecutorService則沒有這個問題。

使用ThreadPoolExecutor建立線程池

以前咱們提到了，不建議使用Executors來建立線程池，而是使用ThreadPoolExecutor進行建立。實際上Executors裏建立的也就是ThreadPoolExecutor的實例，具體的看一下Executors類的源碼就知道了。

接下來用一個例子演示一下如何經過ThreadPoolExecutor來建立線程池，這裏使用7個參數的構造函數，示例代碼以下：

package org.zero.concurrency.demo.example.threadpool;

import lombok.extern.slf4j.Slf4j;
import org.springframework.lang.NonNull;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @program: concurrency-demo
 * @description: ThreadPoolExecutor使用示例
 * @author: 01
 * @create: 2018-10-20 16:35
 **/
@Slf4j
public class ThreadPoolExample6 {
    public static void main(String[] args) {
        // 使用ArrayBlockingQueue做爲其等待隊列
        BlockingQueue<Runnable> blockingQueue = new ArrayBlockingQueue<>(5);
        // 使用自定義的ThreadFactory，目的是設置有意義的的線程名字，方便出錯時回溯
        ThreadFactory namedThreadFactory = new MyThreadFactory("test-thread");

        // 建立線程池
        ThreadPoolExecutor poolExecutor = new ThreadPoolExecutor(
                3, 5, 1, TimeUnit.MINUTES, blockingQueue, namedThreadFactory,
                new ThreadPoolExecutor.AbortPolicy());

        // 執行任務
        poolExecutor.execute(() -> log.info("thread run"));

        // 關閉線程池
        poolExecutor.shutdown();
    }

    private static class MyThreadFactory implements ThreadFactory {
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        private MyThreadFactory(String namePrefix) {
            this.namePrefix = namePrefix + "-";
        }

        @Override
        public Thread newThread(@NonNull Runnable r) {
            Thread t = new Thread(r, namePrefix + threadNumber.getAndIncrement());
            if (t.isDaemon()) {
                t.setDaemon(true);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }
}

線程池的建立先介紹到這，其實大部分的建立方式能夠參考Executors類的源碼，因此這裏就不贅述了。

線程池的合理配置：

  • CPU密集型任務，就須要儘可能壓榨CPU，參考值能夠設置爲NCPU+1，即CPU核心數量+1
  • IO密集型任務，參考值能夠設置爲2*NCPU，即CPU核心數量的2倍

最後須要說一句，線程池雖好但並不是放之四海皆準，咱們應當結合實際業務場景去考慮是否使用線程池。例如當線程池內須要執行的任務很小，小到執行任務的時間和任務調度的時間很接近，這時若使用線程池反而會更慢，由於任務調度和任務管理是須要耗時的。

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