給線程池增加自動擴充,閑時自動回收線程的功能

首先我們給SimpleThreadPool定義三個成員屬性：最小線程數量，活躍線程數量，最大線程數量。

同樣構造方法修改一下：

我們程序默認開啟最小線程數量，當他不夠用的時候我們擴充到活躍線程數量，當活躍線程數量不夠的時候我們擴充到最大線程數量。

那么應該在什么位置控制THREAD_QUEUE呢？當然是由SimpleThreadPool控制了，所以我們用SimpleThreadPool繼承Thread來實現他的run方法：

運行效果如下：

全部代碼如下：

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並發包的知識。