Java学习之自定义线程池

最近面试中，经常会被问到线程池的知识。总感觉自己回答的不够全面。借助资料整理总结一下线程池

Java自带的线程池

要看自定义线程池，需要先看一下线程执行框架 java.util.concurrent.Executors 这个类，这个类扮演的是线程工厂的角色。

通过这个类创建线程池的方法主要有以下四种：

ExecutorService newCachedThreadPool()  返回一个可根据实际情况调整线程个数的线程池，不限制最大线程数数量，若有任务，则创建线程，若无线程则不创建线程，若无线程则在60s后进行线程回收。

ExecutorService newFixedThreadPool (int)  返回一个固定数量的线程池，该方法的线程数始终不变，当有一个线程提交时，若线程池空闲，即立即执行，若没有则会被在一个任务队列中等待有空闲的线程去执行

ExecutorService newSingleThreadExecutor()  返回一个执行器，它在一个单个的线程中依次执行各个任务。

 ScheduledExecutorService newSingleThreadScheduledExecutor() 创建的是一个单线程执行程序，它可安排在给定延迟后运行命令或者定期地执行

 ThreadPoolExecutor

 自带的线程池中线程的创建均是使用了java.util.concurrent.ThreadPoolExecutor类，在这个类的构造器中：

 可以用以下代码表示，主要参数有：

    public ThreadPoolExecutor(int corePoolSize,  //核心线程数
                              int maxNumberSize,  //最大线程数
                              long keepAliveTime, //连接时间数
                              TimeUnit unit,   //时间单位
                              BlockingQueue<Runnable> paramBlockingQueue,  //处理线程队列
                              RejectedExecutionHandler RejectedExecutionHandler  //拒绝处理器
                              ) {

corePoolSize:指定了线程池中的线程数量，它的数量决定了添加的任务是开辟新的线程去执行，还是放到workQueue任务队列中去；

maxNumberPoolSize:指定了线程池中的最大线程数量，这个参数会根据你使用的workQueue任务队列的类型，决定线程池会开辟的最大线程数量；

keepAliveTime:当线程池中空闲线程数量超过corePoolSize时，多余的线程会在多长时间内被销毁；

unit:keepAliveTime的单位

workQueue:任务队列，被添加到线程池中，但尚未被执行的任务；它一般分为直接提交队列、有界任务队列、无界任务队列、优先任务队列几种；

threadFactory:线程工厂，用于创建线程，一般用默认即可；

RejectedExecutionHandler :拒绝策略；当任务太多来不及处理时，如何拒绝任务；

自定义线程池demo

创建一个线程类:

package com.ssc.demo;

public class MyTask implements Runnable {

    private int taskId;
    private String taskName;
    
    public MyTask(int taskId, String taskName){
        this.taskId = taskId;
        this.taskName = taskName;
    }
    
    public int getTaskId() {
        return taskId;
    }

    public void setTaskId(int taskId) {
        this.taskId = taskId;
    }

    public String getTaskName() {
        return taskName;
    }

    public void setTaskName(String taskName) {
        this.taskName = taskName;
    }

    @Override
    public void run() {
        try {
            System.out.println("run taskId =" + this.taskId);
            Thread.sleep(5*1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }        
    }
    
    public String toString(){
        return Integer.toString(this.taskId);
    }

}

插入拒绝处理类：

package com.ssc.demo;

import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;

public class MyRejected implements RejectedExecutionHandler{

    
    public MyRejected(){
    }
    
    @Override
    public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
        System.out.println("自定义处理..");
        System.out.println("当前被拒绝任务为：" + r.toString());
    }

}

使用有界队列自定义线程池时：

package com.ssc.demo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;



public class UseThreadPoolExecutor1 {


    public static void main(String[] args) {
        /**
         * 在使用有界队列时，若有新的任务需要执行，如果线程池实际线程数小于corePoolSize，则优先创建线程，
         * 若大于corePoolSize，则会将任务加入队列，
         * 若队列已满，则在总线程数不大于maximumPoolSize的前提下，创建新的线程，
         * 若线程数大于maximumPoolSize，则执行拒绝策略。或其他自定义方式。
         * 
         */    
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                1,                 //coreSize
                2,                 //MaxSize
                60,             //60
                TimeUnit.SECONDS, 
                new ArrayBlockingQueue<Runnable>(3)            //指定一种队列 （有界队列）
                //new LinkedBlockingQueue<Runnable>()
                , new MyRejected()
                //, new DiscardOldestPolicy()
                );
        
        MyTask mt1 = new MyTask(1, "任务1");
        MyTask mt2 = new MyTask(2, "任务2");
        MyTask mt3 = new MyTask(3, "任务3");
        MyTask mt4 = new MyTask(4, "任务4");
        MyTask mt5 = new MyTask(5, "任务5");
        MyTask mt6 = new MyTask(6, "任务6");
        
        pool.execute(mt1);
        pool.execute(mt2);
        pool.execute(mt3);
        pool.execute(mt4);
        pool.execute(mt5);
        pool.execute(mt6);
        
        pool.shutdown();
        
    }
}

运行该类时。效果如下

 使用无界队列自定义线程池时

package com.ssc.demo;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class UseThreadPoolExecutor2 implements Runnable{

    private static AtomicInteger count = new AtomicInteger(0);
    
    @Override
    public void run() {
        try {
            int temp = count.incrementAndGet();
            System.out.println("任务" + temp);
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    
    public static void main(String[] args) throws Exception{
        //System.out.println(Runtime.getRuntime().availableProcessors());
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<Runnable>();
        ExecutorService executor  = new ThreadPoolExecutor(
                    5,         //core
                    10,     //max
                    120L,     //2fenzhong
                    TimeUnit.SECONDS,
                    queue);
        
        for(int i = 0 ; i < 20; i++){
            executor.execute(new UseThreadPoolExecutor2());
        }
        Thread.sleep(1000);
        System.out.println("queue size:" + queue.size());        //10
        Thread.sleep(2000);
    }


}

允许结果为：

该项目主要参考对  北京尚学堂的白鹤翔视频

java线程池ThreadPoolExecutor类使用详解