26.給線程池增加自動擴充,閑時自動回收線程的功能 · 技術筆記

# 給線程池增加自動擴充,閑時自動回收線程的功能首先我們給SimpleThreadPool定義三個成員屬性：最小線程數量，活躍線程數量，最大線程數量。 ![](https://img.kancloud.cn/2e/76/2e76147298519aad04b7177716ea44a3_640x312.png) ![](https://img.kancloud.cn/de/d1/ded1b47584f724f0278938fb4458a22d_592x330.png) 同樣構造方法修改一下： ![](https://img.kancloud.cn/e4/3c/e43c1dfc04c4c84fb5b0df0677d25613_966x414.png) 我們程序默認開啟最小線程數量，當他不夠用的時候我們擴充到活躍線程數量，當活躍線程數量不夠的時候我們擴充到最大線程數量。那么應該在什么位置控制THREAD_QUEUE呢？當然是由SimpleThreadPool控制了，所以我們用SimpleThreadPool繼承Thread來實現他的run方法： ![](https://img.kancloud.cn/aa/3f/aa3f58da8f6ad9749cd72841ac05d9ab_936x810.png) 運行效果如下： ![](https://img.kancloud.cn/1b/fc/1bfc02fb1badfe86ceba57f6d756744f_1340x369.gif) 全部代碼如下： ```java package com.thread.thread22; import java.util.ArrayList; import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.IntStream; /** * 自定義簡單線程池 * 任務隊列、拒絕策略 */ public class SimpleThreadPool extends Thread { /** * 最小線程數量 */ private int min; /** * 活躍線程數量 */ private int active; /** * 最大線程數量 */ private int max; /** * 傳參線程池大小(當前線程池中線程的數量) */ private int size; /** * 線程隊列的大小 */ private final int queueSize; /** * 拒絕策略 */ private final DiscardPolicy discardPolicy; /** * 線程池是否被銷毀 */ private volatile boolean destroy = false; /** * 默認最小線程數量 */ private static final int DEFAULT_MIN_SIZE = 4; /** * 默認活躍線程數量 */ private static final int DEFAULT_ACTIVE_SIZE = 8; /** * 默認最大線程數量 */ private static final int DEFAULT_MAX_SIZE = 12; /** * 默認線程隊列的大小 */ private static final int DEFAULT_TASK_QUEUE_SIZE = 2000; /** * 默認拒絕策略-拋出異常 */ public static final DiscardPolicy DEFAULT_DISCARD_POLICY = () -> { throw new DiscardException("Discard This Task."); }; /** * 線程名自增序號 */ private static AtomicInteger sequence = new AtomicInteger(); /** * 線程名前綴 */ private static final String THREAD_PREFIX = "SIMPLE_THREAD_POOL-"; /** * 線程組 */ private static final ThreadGroup GROUP = new ThreadGroup("Pool_Group"); /** * 任務隊列 */ private static final LinkedList<Runnable> TASK_QUEUE = new LinkedList<>(); /** * 線程隊列 */ private static final List<WorkerTask> THREAD_QUEUE = new ArrayList<>(); public SimpleThreadPool() { this(DEFAULT_MIN_SIZE, DEFAULT_ACTIVE_SIZE, DEFAULT_MAX_SIZE, DEFAULT_TASK_QUEUE_SIZE, DEFAULT_DISCARD_POLICY); } /** * 構造方法 * * @param min 最小線程數量 * @param active 活躍線程數量 * @param max 最大線程數量 * @param queueSize 線程隊列的大小 * @param discardPolicy 拒絕策略 */ public SimpleThreadPool(int min, int active, int max, int queueSize, DiscardPolicy discardPolicy) { this.min = min; this.active = active; this.max = max; this.queueSize = queueSize; this.discardPolicy = discardPolicy; init(); } @Override public void run() { while (!destroy) { System.out.printf("Pool#Min:%d,Active:%d,Max:%d,Current:%d,QueueSize:%d\n", this.min, this.active, this.max, this.size, TASK_QUEUE.size()); try { Thread.sleep(5_000L); // 任務隊列數量過大，線程隊列的數量擴充到active if (size < active && active < TASK_QUEUE.size()) { IntStream.range(size, active).forEach(i -> createWorkTask()); size = active; System.out.println("The Pool incremented to active."); } // 任務隊列數量過大，線程隊列的數量擴充到max else if (size < max && max < TASK_QUEUE.size()) { IntStream.range(active, max).forEach(i -> createWorkTask()); size = max; System.out.println("The Pool incremented to max."); } // 防止新插入任務隊列的任務使用線程隊列中的線程執行任務。 synchronized (THREAD_QUEUE) { // 任務隊列數量為空，線程隊列的數量減少到active if (size > active && TASK_QUEUE.isEmpty()) { System.out.println("=========Reduce========="); int releaseSize = size - active; for (Iterator<WorkerTask> it = THREAD_QUEUE.iterator(); it.hasNext(); ) { if (releaseSize <= 0) { break; } WorkerTask task = it.next(); task.interrupt(); task.close(); it.remove(); releaseSize--; } size = active; } } } catch (InterruptedException e) { e.printStackTrace(); } } } private void init() { IntStream.range(0, this.min).forEach(i -> createWorkTask()); this.size = this.min; this.start(); } private void createWorkTask() { WorkerTask task = new WorkerTask(GROUP, THREAD_PREFIX + (sequence.getAndIncrement())); task.start(); THREAD_QUEUE.add(task); } /** * 提交任務 * * @param runnable 任務 */ public void submit(Runnable runnable) { // 如果線程池被銷毀，不能添加任務到任務隊列 if (destroy) { throw new IllegalThreadStateException("The thread pool already destroy and not allow submit task."); } synchronized (TASK_QUEUE) { // 任務隊列數量超過閾值，則執行拒絕策略 if (TASK_QUEUE.size() > queueSize) { discardPolicy.discard(); } TASK_QUEUE.addLast(runnable); TASK_QUEUE.notifyAll(); } } /** * 關閉線程池中的線程 * * @throws InterruptedException */ public void shutdown() throws InterruptedException { // 當任務隊列還有任務，則稍微等待。 while (!TASK_QUEUE.isEmpty()) { Thread.sleep(50); } // 判斷線程中有沒有正在運行的任務 synchronized (THREAD_QUEUE) { int initVal = THREAD_QUEUE.size(); while (initVal > 0) { for (WorkerTask task : THREAD_QUEUE) { if (task.getTaskState() == TaskState.BLOCKED) { task.interrupt(); task.close(); initVal--; } else { Thread.sleep(10); } } } } System.out.println("Group active count is " + GROUP.activeCount()); this.destroy = true; System.out.println("The thread pool disposed."); } public boolean isDestroy() { return destroy; } public int getMin() { return min; } public int getMax() { return max; } public int getActive() { return active; } public int getSize() { return size; } public int getQueueSize() { return queueSize; } /** * 封裝任務類 */ private static class WorkerTask extends Thread { private volatile TaskState taskState = TaskState.FREE; public TaskState getTaskState() { return this.taskState; } public WorkerTask(ThreadGroup group, String name) { super(group, name); } @Override public void run() { OUTER: while (this.taskState != TaskState.DEAD) { Runnable runnable; synchronized (TASK_QUEUE) { // 任務隊列為空，則進入阻塞狀態 while (TASK_QUEUE.isEmpty()) { try { this.taskState = TaskState.BLOCKED; TASK_QUEUE.wait(); } catch (InterruptedException e) { System.out.println("Closed."); break OUTER; } } // 任務隊列不為空，取出任務 runnable = TASK_QUEUE.removeFirst(); } // 任務不為空，則執行任務 if (runnable != null) { this.taskState = TaskState.RUNNING; runnable.run(); this.taskState = TaskState.FREE; } } } public void close() { this.taskState = TaskState.DEAD; } } /** * 線程狀態 */ private enum TaskState { FREE, RUNNING, BLOCKED, DEAD } /** * 拒絕策略 */ public interface DiscardPolicy { void discard() throws DiscardException; } /** * 拒絕策略異常 */ public static class DiscardException extends RuntimeException { public DiscardException(String message) { super(message); } } public static void main(String[] args) throws InterruptedException { SimpleThreadPool threadPool = new SimpleThreadPool(); /*SimpleThreadPool threadPool = new SimpleThreadPool(6, 10, SimpleThreadPool.DEFAULT_DISCARD_POLICY);*/ IntStream.rangeClosed(1, 40).forEach(i -> { threadPool.submit(() -> { System.out.printf("The runnable %d be serviced by %s start.\n", i, Thread.currentThread().getName()); try { Thread.sleep(3_000L); } catch (InterruptedException e) { e.printStackTrace(); } System.out.printf("The runnable %d be serviced by %s finished.\n", i, Thread.currentThread().getName()); }); }); Thread.sleep(30_000L); threadPool.shutdown();// 關閉線程池中的線程 /*threadPool.submit(() -> System.out.println("線程池被銷毀，不能添加任務到任務隊列"));*/ } } ``` java的多線程基礎就暫時學完了，后面還要繼續學習多線程的設計模式以及jdk并發包的知識。