java如何自定义一个线程池

Posted 2021-03-02 Mr.lang?

java线程池的一些简单功能，后续会更新，代码不多，很好理解

package com.rbac.thread;


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;



/**
 * 自定义线程池
 * 
 * @author xl,lang
 *
 */
public class MyExecutorService {
    // 初始化线程
    protected List<InitThread> initThreads;

    // 执行任务列表
    protected BlockingQueue<Runnable> taskQueues;

    // 线程执行状态
    protected volatile boolean threadState = true;

    /*
     * // 当前线程的活跃数 public AtomicInteger activeCount; public Lock lock=new
     * ReentrantLock(); // 最小活跃数 public int threadMinSize = 0; // 最大线程数 public
     * int threadMaxSize = 0; // 初始话线程数 public int threadInitSize = 0;
     */

    /**
     * 线程初始化方法
     * 
     * @param threadMaxSize
     * @param threadInitSize
     * @param taskQueueSize
     */
    /*
     * public MyExecutorService(int threadMaxSize, int threadMinSize, int
     * threadInitSize, int taskQueueSize) { this.taskQueues = new
     * LinkedBlockingDeque<Runnable>(taskQueueSize); if (threadInitSize > 0 &&
     * threadInitSize < threadMaxSize) { this.initThreads = new ArrayList<>();
     * for (int i = 0; i < threadInitSize; i++) { InitThread init = new
     * InitThread(); init.start(); initThreads.add(init); }
     * 
     * }
     * 
     * }
     */

    /**
     * 线程初始化方法
     * 
     * @param threadMaxSize
     * @param threadInitSize
     * @param taskQueueSize
     */
    public MyExecutorService(int threadInitSize, int taskQueueSize) {
        this.taskQueues = new LinkedBlockingDeque<Runnable>(taskQueueSize);
        if (threadInitSize > 0) {
            this.initThreads = new ArrayList<>();
            for (int i = 0; i < threadInitSize; i++) {
                InitThread init = new InitThread();
                init.start();
                initThreads.add(init);
            }

        }

    }

    // 添加任务
    public boolean exeute(Runnable task) throws InterruptedException {

        return this.taskQueues.offer(task, 60, TimeUnit.SECONDS);
    }

    /**
     * 初始化线程
     * 
     * @author xl,lang
     *
     */
    class InitThread extends Thread {

        @Override
        public void run() {

            while (threadState || taskQueues.size() > 0) {
                Runnable runable = taskQueues.poll();
                if (null != runable) {
                    runable.run();
                }

            }

        }

    }

    // 关闭线程池
    public void shutdown() {
        this.threadState = false;
    }

    public static void main(String[] args) {

        MyExecutorService es = new MyExecutorService(2, 5);
        for (int i = 0; i < 100; i++) {
            final int a = i;
            try {
                es.exeute(new Runnable() {

                    @Override
                    public void run() {
                        System.out.println(Thread.currentThread().getName() + "执行" + a);

                    }
                });
            } catch (InterruptedException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
        }
        es.shutdown();

    }
}