Java实现自定义线程池

ThreadPool

public interface ThreadPool {

    void execute(Runnable runnable);

    void shutdown();

    int getInitSize();

    int getMaxSize();

    int getCoreSize();

    int getQueueSize();

    int getActiveCount();

    boolean isShutdown();
}

RunnableQueue

public interface RunnableQueue {

    void offer(Runnable runnable);

    Runnable take() throws InterruptedException;

    int size();
}

ThreadFactory

public interface ThreadFactory {

    Thread createThread(Runnable r);

}

DennyPolicy

public interface DennyPolicy {

    void reject(Runnable r, ThreadPool pool);

    class DiscardPolicy implements DennyPolicy {

        @Override
        public void reject(Runnable r, ThreadPool pool) {
            // do nothing
        }
    }

    class AbortPolicy implements DennyPolicy {

        @Override
        public void reject(Runnable r, ThreadPool pool) {
            throw new RejectedExecutionException();
        }
    }

    class RunnerDenyPolicy implements DennyPolicy {

        @Override
        public void reject(Runnable r, ThreadPool pool) {
            if (!pool.isShutdown()) {
                r.run();
            }
        }
    }

}

RunnableDenyException

public class RunnableDenyException extends RuntimeException {

    public RunnableDenyException(String message) {
        super(message);
    }

}

InternalTask

public class InternalTask implements Runnable{

    private final RunnableQueue runnableQueue;

    private volatile boolean running = true;

    public InternalTask(RunnableQueue runnableQueue) {
        this.runnableQueue = runnableQueue;
    }


    @Override
    public void run() {
        while (running && !Thread.currentThread().isInterrupted()) {
            try {
                Runnable task = runnableQueue.take();
                task.run();
            } catch (InterruptedException e) {
                running = false;
                break;
            }
        }
    }

    public void stop() {
        running = false;
    }

}

LinkedRunnableQueue

public class LinkedRunnableQueue implements RunnableQueue {

    private final int limit;

    private final DennyPolicy dennyPolicy;

    private final LinkedList<Runnable> runnableList = new LinkedList<>();

    private final ThreadPool threadPool;

    public LinkedRunnableQueue(final int limit, final DennyPolicy dennyPolicy, ThreadPool threadPool) {
        this.limit = limit;
        this.dennyPolicy = dennyPolicy;
        this.threadPool = threadPool;
    }

    @Override
    public void offer(Runnable runnable) {
        synchronized (runnableList) {
            if (runnableList.size() >= limit) {
                dennyPolicy.reject(runnable, threadPool);
            } else {
                runnableList.add(runnable);
                runnableList.notify();
            }
        }
    }

    @Override
    public Runnable take() throws InterruptedException {
        synchronized (runnableList) {
            while (runnableList.isEmpty()) {
                try {
                    runnableList.wait();
                } catch (InterruptedException e) {
                    throw e;
                }
            }
        }
        return runnableList.removeFirst();
    }

    @Override
    public int size() {
        synchronized (runnableList) {
            return runnableList.size();
        }
    }
}

BasicThreadPool

public class BasicThreadPool extends Thread implements ThreadPool {

    private final int initSize;
    private final int maxSize;
    private final int coreSize;
    private int activeCount;

    private final ThreadFactory threadFactory;

    private final RunnableQueue runnableQueue;

    private volatile boolean isShutdown = false;

    private final Queue<ThreadTask> threadQueue = new ArrayDeque<ThreadTask>();

    private final static DennyPolicy DEFAULT_DENNY_POLICY = new DennyPolicy.DiscardPolicy();

    private final static ThreadFactory DEFAULT_THREAD_FACTORY = new DefaultThreadFactory();

    private final long keepAliveTime;

    private final TimeUnit timeUnit;

    public BasicThreadPool(int initSize, int maxSize, int coreSize, int queueSize) {
        this(initSize, maxSize, coreSize, DEFAULT_THREAD_FACTORY, queueSize, DEFAULT_DENNY_POLICY, 10, TimeUnit.SECONDS);
    }

    public BasicThreadPool(int initSize, int maxSize, int coreSize, ThreadFactory threadFactory, int queueSize, DennyPolicy dennyPolicy, long keepAliveTime, TimeUnit unit) {
        this.initSize = initSize;
        this.maxSize = maxSize;
        this.coreSize = coreSize;
        this.threadFactory = threadFactory;
        this.runnableQueue = new LinkedRunnableQueue(queueSize, dennyPolicy, this);
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = unit;
        this.init();
    }

    private void init() {
        start();
        for (int i = 0; i < initSize; i++) {
            newThread();
        }
    }

    private void newThread() {
        InternalTask internalTask = new InternalTask(runnableQueue);
        Thread thread = this.threadFactory.createThread(internalTask);
        ThreadTask threadTask = new ThreadTask(thread, internalTask);
        threadQueue.offer(threadTask);
        this.activeCount++;
        thread.start();
    }

    private void removeThread() {
        ThreadTask threadTask = threadQueue.remove();
        threadTask.internalTask.stop();
        this.activeCount--;
    }

    @Override
    public void run() {
        while (!isShutdown && !isInterrupted()) {
            try {
                timeUnit.sleep(keepAliveTime);
            } catch (InterruptedException e) {
                isShutdown = true;
                break;
            }

            synchronized (this) {
                if (isShutdown) {
                    break;
                }

                if (runnableQueue.size() > 0 && activeCount < coreSize) {
                    for (int i = initSize; i < coreSize; i++) {
                        newThread();
                    }
                    continue;
                }

                if (runnableQueue.size() > 0 && activeCount < maxSize) {
                    for (int i = coreSize; i < maxSize; i++) {
                        newThread();
                    }
                }

                if (runnableQueue.size() == 0 && activeCount > coreSize) {
                    for (int i = coreSize; i < activeCount; i++) {
                        removeThread();
                    }
                }

            }
        }
    }

    private static class ThreadTask {

        Thread thread;
        InternalTask internalTask;

        public ThreadTask(Thread thread, InternalTask internalTask) {
            this.thread = thread;
            this.internalTask = internalTask;
        }
    }

    @Override
    public void execute(Runnable runnable) {
        if (this.isShutdown) {
            throw new IllegalStateException("ThreadPool is closed");
        }
        this.runnableQueue.offer(runnable);
    }

    @Override
    public void shutdown() {
        synchronized (this) {
            if (isShutdown) {
                return;
            }

            isShutdown = true;
            threadQueue.forEach(threadTask -> {
                threadTask.internalTask.stop();
                threadTask.thread.interrupt();
            });
            this.interrupt();
        }
    }

    @Override
    public int getInitSize() {
        if (isShutdown) {
            throw new IllegalStateException("ThreadPool is closed");
        }
        return this.initSize;
    }

    @Override
    public int getMaxSize() {
        if (isShutdown) {
            throw new IllegalStateException("ThreadPool is closed");
        }
        return this.maxSize;
    }

    @Override
    public int getCoreSize() {
        if (isShutdown) {
            throw new IllegalStateException("ThreadPool is closed");
        }
        return this.coreSize;
    }

    @Override
    public int getQueueSize() {
        if (isShutdown) {
            throw new IllegalStateException("ThreadPool is closed");
        }
        return runnableQueue.size();
    }

    @Override
    public int getActiveCount() {
        synchronized (this) {
            return activeCount;
        }
    }

    @Override
    public boolean isShutdown() {
        return this.isShutdown;
    }

    private static class DefaultThreadFactory implements ThreadFactory {
        private static final AtomicInteger GROUP_COUNTER = new AtomicInteger(1);

        private static final ThreadGroup group = new ThreadGroup("MyThreadPool" + GROUP_COUNTER.getAndDecrement());

        private static final AtomicInteger COUNTER = new AtomicInteger(1);

        @Override
        public Thread createThread(Runnable runnable) {
            return new Thread(group, runnable, "thread-pool-" + COUNTER.getAndDecrement());
        }
    }

}

ThreadPoolTest

public class ThreadPoolTest {

    public static void main(String[] args) throws InterruptedException {

        final BasicThreadPool threadPool = new BasicThreadPool(2, 6, 4, 1000);

        for (int i = 0; i < 20; i++) {
            threadPool.execute(() -> {
                try {
                    TimeUnit.SECONDS.sleep(10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }

        for (; ; ) {
            System.out.println("getActiveCount() = " + threadPool.getActiveCount());
            System.out.println("getQueueSize() = " + threadPool.getQueueSize());
            System.out.println("getCoreSize() = " + threadPool.getCoreSize());
            System.out.println("getMaxsize() = " + threadPool.getMaxSize());
            System.out.println("=============================================");
            TimeUnit.SECONDS.sleep(5);
        }
    }
}

运行结果

getActiveCount() = 2
getQueueSize() = 18
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 2
getQueueSize() = 18
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 4
getQueueSize() = 14
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 4
getQueueSize() = 14
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 6
getQueueSize() = 8
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 6
getQueueSize() = 8
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 6
getQueueSize() = 2
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 6
getQueueSize() = 2
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 5
getQueueSize() = 0
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 5
getQueueSize() = 0
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 4
getQueueSize() = 0
getCoreSize() = 4
getMaxsize() = 6
=============================================
getActiveCount() = 4
getQueueSize() = 0
getCoreSize() = 4
getMaxsize() = 6
=============================================