并发线程基础第八篇

线程池

自定义线程池

步骤1：自定义拒绝策略接口

步骤2：自定义任务队列

步骤3：自定义线程池

步骤 4：测试

ThreadPoolExecutor

线程池状态

构造方法

工作方式

newFixedThreadPool

newCachedThreadPool

newSingleThreadExecutor

提交任务

1.执行任务

2.提交任务task，用返回值Future获得任务执行结果

3.提交tasks中所有的任务

4.提交 tasks 中所有任务，带超时时间

5.提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消

6.提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间

关闭线程池

shutdown

shutdownNow

任务调度线程池

Timer

ScheduledExecutorService

Tomcat 线程池

Fork/Join

概念

使用

线程池

自定义线程池

步骤1：自定义拒绝策略接口

@FunctionalInterface // 拒绝策略
interface RejectPolicy<T> {
 void reject(BlockingQueue<T> queue, T task);
}

步骤2：自定义任务队列

class BlockingQueue<T>{
    //1.任务队列
    private Deque<T> queue = new ArrayDeque<>();

    //2.锁
    private ReentrantLock lock = new ReentrantLock();

    //生产者条件变量
    private Condition fullWaitSet = lock.newCondition();
    //4.消费者条件变量
    private Condition emptyWaitSet = lock.newCondition();

    //5.容量
    private int capcity;

    public BlockingQueue(int capcity){
        this.capcity = capcity;
    }
    //超时的阻塞添加
    public T poll(long timeout, TimeUnit unit){
        lock.lock();
        try{
            //将时间统一转化为纳秒
            long nanos = unit.toNanos(timeout);
            while(queue.isEmpty()){
                try {
                    if(nanos<=0){
                        return null;
                    }
                    //awaitNanos()方法，返回的时间是一个剩余时间，把返回的时间重新赋值给它，防止虚假唤醒，从头再等
                    nanos = emptyWaitSet.awaitNanos(nanos);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }

    //阻塞获取
    public T take(){
        //获取任务的时候需要保证线程安全，需要加一把锁
        lock.lock();
        try{
            while(queue.isEmpty()){
                try {
                    emptyWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            T t = queue.removeFirst();
            fullWaitSet.signal();
            return t;
        }finally {
            lock.unlock();
        }
    }
    //阻塞添加
    public void put (T element){
        lock.lock();
        try{
            while(queue.size()==capcity){
                try {
                    fullWaitSet.await();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            queue.addLast(element);
            emptyWaitSet.signal();
        }finally {
            lock.unlock();
        }
    }
    //获取当前队列的大小
    public int size(){
        lock.lock();
        try {
            return queue.size();
        }finally {
            lock.unlock();
        }
    }
}

步骤3：自定义线程池

class ThreadPool{
    //阻塞队列
    private BlockingQueue<Runnable> taskQueue;
    //线程集合
    private HashSet<Worker> workers = new HashSet<>();
    //核心线程数
    private int coreSize;
    //获取任务的超时时间
    private long timeout;
    //时间单位
    private TimeUnit timeUnit;
    //执行任务
    public void excute(Runnable task){
        //当任务数没有超过coreSize时，直接交给worker执行
        //当任务数超过coreSize时候，加入任务队列暂存
        synchronized (workers){
            if(workers.size()<coreSize){
                Worker worker = new Worker(task);
                log.debug("新增 worker{},任务对象{}",worker,task);
                workers.add(worker);
                worker.start();
            }else {
                log.debug("加入任务队列{}",task);
                taskQueue.put(task);
            }
        }
    }
    //构造方法
    public ThreadPool(int coreSize, long timeout, TimeUnit timeUnit,int queueCapcity) {
        this.coreSize = coreSize;
        this.timeout = timeout;
        this.timeUnit = timeUnit;
        this.taskQueue = new BlockingQueue<>(queueCapcity);
    }

    class Worker extends Thread{
        private Runnable task;
        public Worker(Runnable task){
            this.task=task;
        }
        @Override
        public void run(){
            //执行任务
            //当task不为空，执行任务
            //当task执行完毕，再接着从任务队列获取任务并执行
//            while (task !=null ||(task = taskQueue.take()) !=null){
            while (task !=null ||(task = taskQueue.poll(timeout,timeUnit)) !=null){
                try {
                    log.debug("正在执行...{}",task);
                    task.run();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    task = null;
                }
            }
            synchronized (workers){
                log.debug("worker 被移除{}",this);
                workers.remove(this);
            }
        }
    }
}

步骤 4：测试

public class Test {
    public static void main(String[] args) {
        ThreadPool threadPool = new ThreadPool(2, 1000, TimeUnit.MILLISECONDS, 10);
        for (int i = 0; i < 5; i++) {
            int j =i;
            threadPool.excute(()->{
                log.debug("{}",j);
            });
        }
    }
}

以上是一个基本的不借助不借助jdk提供线程池框架的简单例子，可以自行多练习几遍，有助于理解jdk提供线程框架的底层。

ThreadPoolExecutor

线程池状态

ThreadPoolExecutor使用int的高3位来表示线程池状态，低29位表示线程数量

状态名	高3位	接收新任务	处理阻塞队列任务	说明
RUNNING	111	Y	Y
SHUTDOWN	000	N	Y	不会接收新任务，但是会处理阻塞队列剩余任务
STOP	001	N	N	会中断正在执行的任务，并抛弃阻塞队列任务
TIDYING	010	-	-	任务执行完毕，活动线程为0即将进入终结
TERMINATED	011	-	-	终结状态

从数字上比较，TERMINATED > TIDYING > STOP > SHUTDOWN > RUNNING

这些信息存储在一个原子变量ctl中，目的是将线程状态与线程个数合二为一，这样就可以用一次cas原子操作进行赋值

// c 为旧值， ctlOf 返回结果为新值
ctl.compareAndSet(c, ctlOf(targetState, workerCountOf(c))));
 
// rs 为高 3 位代表线程池状态， wc 为低 29 位代表线程个数，ctl 是合并它们
private static int ctlOf(int rs, int wc) { return rs | wc; }

构造方法

public ThreadPoolExecutor(
            int corePoolSize,
            int maximumPoolSize,
            long keepAliveTime,
            TimeUnit unit,
            BlockingQueue<Runnable> workQueue,
            ThreadFactory threadFactory,
            RejectedExecutionHandler handler
    )

corePoolSize核心线程数目（最多保留的线程数）
maximumPoolSize最大线程数目
keepAliveTime生存时间-针对救急线程
unit 时间单位-针对救急线程
workQueue阻塞队列
threadFactory线程工厂-可以为线程创建的时候起名字
handler拒绝策略

工作方式

1.线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务

2.当线程数达到corePoolSize的时候，并没有线程空闲，再来新任务的时候会加入任务队列workQueue队列排队，直到有空闲的线程

3.如果队列选择了有界队列，那么任务超过了队列大小时，会创建maximumPoolSize-corePoolSize的数量的线程来救急。

4.如果线程达到了maximumPoolSize的时候，仍有新任务这时候会执行拒绝策略。拒绝策略jdk提供了4种实现，其他著名框架也提供了实现

AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
CallerRunsPolicy 让调用者运行任务
DiscardPolicy 放弃本次任务
DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
Netty 的实现，是创建一个新线程来执行任务
ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略

5.当高峰过去后，超过corePoolSize的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime和unit控制

根据这个构造方法，JDK Executors 类中提供了众多工厂方法来创建各种用途的线程池

newFixedThreadPool

public static ExecutorService newFixedThreadPool(int nThreads) {
     return new ThreadPoolExecutor(nThreads, nThreads,
                                     0L, TimeUnit.MILLISECONDS,
                                     new LinkedBlockingQueue<Runnable>());
}

特点

核心线程数==最大线程数（没有救急线程），因此也无需超时时间
阻塞队列是无界的，可以放任意数据的任务
适用于任务量已知，相对耗时的任务

newCachedThreadPool

public static ExecutorService newCachedThreadPool() {
     return new ThreadPoolExecutor(  0, Integer.MAX_VALUE,
                                     60L, TimeUnit.SECONDS,
                                     new SynchronousQueue<Runnable>());
}

特点

核心线程数是0，最大线程数是Integer.MAX_VALUE,救急线程的生存时间为60s，意味着全部都是救急线程（60s后可以回收），救急线程可以无限创建
队列采用了SynchronousQueue实现特点是，它没有容量，没有线程来取是放不进去的（一手交钱，一手交货）
整个线程池表现为线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。适合任务数比较密集，但每个任务执行时间较短的情况

示例

public class Test2 {
    public static void main(String[] args) {
        SynchronousQueue<Integer> integers = new SynchronousQueue<>();
        new Thread(()->{

            try {
                log.debug("putting{}",1);
                integers.put(1);
                log.debug("putted{}",1);
                log.debug("putting{}",2);
                integers.put(2);
                log.debug("putted{}",2);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
    }
}
07:37:56.496 [Thread-0] DEBUG com.example.com.yunlong.test3.Test2 - putting1

从上面示例结果可以看出来，integers队列，把1放进去之后，如果没有来取的，会直接阻塞到放入元素行，不再执行下去。

newSingleThreadExecutor

public static ExecutorService newSingleThreadExecutor() {
 return new FinalizableDelegatedExecutorService
             (new ThreadPoolExecutor(1, 1,
                                     0L, TimeUnit.MILLISECONDS,
                                     new LinkedBlockingQueue<Runnable>()));
}

使用场景

希望多个任务排队执行。线程数固定为1，任务数多余1时，会放入无界队列排队。任务执行完毕，这唯一的线程也不会被释放

区别

自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作

Executors.newSingleThreadExecutor()线程个数始终为1，不能修改

FinalizableDelegatedExecutorService 应用的是装饰器模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法

Executors.newFixedThreadPool(1) 初始时为1，以后还可以修改

对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

提交任务

1.执行任务

void execute(Runnable command);

2.提交任务task，用返回值Future获得任务执行结果

<T> Future<T> submit(Callable<T> task);

示例

public class Test4 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(2);
        Future<String> f = executorService.submit(() -> {
            log.debug("running");
            Thread.sleep(1000);
            return "1";
        });
        //用Future对象，调用get()方法拿到结果，如果拿不到结果就在这阻塞住
        log.debug("{}",f.get());
    }
}

总结：Runnable接口中的唯一抽象方法run()方法没有返回值，Callable接口中的唯一抽象方法call()方法有返回值。

@FunctionalInterface
public interface Runnable {
    // 这个run()方法返回值为void，没有受检异常的异常声明，出现异常只能内部捕获
    public abstract void run();
}

@FunctionalInterface
public interface Callable<V> {
	// 这个call()方法有返回值，且声明了受检异常，可以直接抛出Exception异常
    V call() throws Exception;
}

3.提交tasks中所有的任务

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
 throws InterruptedException;

4.提交 tasks 中所有任务，带超时时间

<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,
 long timeout, TimeUnit unit)
 throws InterruptedException;

5.提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消

<T> T invokeAny(Collection<? extends Callable<T>> tasks)
 throws InterruptedException, ExecutionException;

6.提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间

<T> T invokeAny(Collection<? extends Callable<T>> tasks,
 long timeout, TimeUnit unit)
 throws InterruptedException, ExecutionException, TimeoutException;

关闭线程池

shutdown

/*
线程池状态变为 SHUTDOWN
 - 不会接收新任务
 - 但已提交任务会执行完
 - 此方法不会阻塞调用线程的执行
*/
void shutdown();

public void shutdown() {
 final ReentrantLock mainLock = this.mainLock;
 mainLock.lock();
 try {
 checkShutdownAccess();
 // 修改线程池状态
 advanceRunState(SHUTDOWN);
 // 仅会打断空闲线程
 interruptIdleWorkers();
 onShutdown(); // 扩展点 ScheduledThreadPoolExecutor
 } finally {
 mainLock.unlock();
 }
 // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
 tryTerminate();
}

shutdownNow

/*
线程池状态变为 STOP
 - 不会接收新任务
 - 会将队列中的任务返回
 - 并用 interrupt 的方式中断正在执行的任务
*/
List<Runnable> shutdownNow();

public List<Runnable> shutdownNow() {

     List<Runnable> tasks;
     final ReentrantLock mainLock = this.mainLock;
     mainLock.lock();
     try {
     checkShutdownAccess();
     // 修改线程池状态
     advanceRunState(STOP);
     // 打断所有线程
     interruptWorkers();
     // 获取队列中剩余任务
     tasks = drainQueue();
     } finally {
     mainLock.unlock();
     }
     // 尝试终结
     tryTerminate();
     return tasks;
}

任务调度线程池

Timer

在任务调度线程池功能加入之前，可以使用java.util.Timer来实现定时功能，Timer的优点在于简单易用，但由于所有任务都是由同一线程来调度，因此所有任务都是串行执行的，同一时间只能有一个任务在执行，前一个任务的延迟或异常都将会影响到之后的任务。

示例

public static void main(String[] args) throws ExecutionException, InterruptedException {
        Timer timer = new Timer();
        TimerTask task1 = new TimerTask() {
            @Override
            public void run() {
                log.debug("task1");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        };
        TimerTask task2 = new TimerTask() {
            @Override
            public void run() {
                log.debug("task2");
            }
        };
        //可以看到task1延迟2秒，此时线程会等待task1执行完，才能执行后边的任务
        timer.schedule(task1,1000);
        timer.schedule(task2,1000);
    }

08:04:04.329 [Timer-0] DEBUG com.example.com.yunlong.test3.Test4 - task1
08:04:06.366 [Timer-0] DEBUG com.example.com.yunlong.test3.Test4 - task2

ScheduledExecutorService

使用ScheduledExecutorService改写：

public class Test4 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
        scheduledExecutorService.schedule(()->{
            log.debug("task1");
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },1,TimeUnit.SECONDS);

        scheduledExecutorService.schedule(()->{
            log.debug("task2");
        },1,TimeUnit.SECONDS);

    }
}
08:11:38.793 [pool-1-thread-1] DEBUG com.example.com.yunlong.test3.Test4 - task1
08:11:38.793 [pool-1-thread-2] DEBUG com.example.com.yunlong.test3.Test4 - task2

可以看到前一个任务的执行延迟并不会影响到另外的线程。

Tomcat 线程池

为什么jdk提供了线程池，tomcat还要自定义线程池，是因为jdk提供的线程池是cpu类型的(cpu计算类型的任务处理较快，处理完了可以去queue再取task)，而tomcat 处理的请求大多数io相关的，如果核心线程满了，就入队，那io请求就会被阻塞，所以tomcat是线程个数到最大线程数之后，才会进队列，这个和jdk的有点区别

LimitLatch 用来限流，可以控制最大连接个数，类似 J.U.C 中的 Semaphore 后面再讲
Acceptor 只负责【接收新的 socket 连接】
Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
Executor 线程池中的工作线程最终负责【处理请求】

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同

如果总线程数达到 maximumPoolSize

这时不会立刻抛 RejectedExecutionException 异常而是再次尝试将任务放入队列，
如果还失败，才抛出 RejectedExecutionException 异常

public void execute(Runnable command,long timeout,TimeUnit unit){
        submittedCount.incrementAndGet();
        try{
        super.execute(command);
            }catch(RejectedExecutionException rx){
                if(super.getQueue()instanceof TaskQueue){
                    final TaskQueue queue=(TaskQueue)super.getQueue();
                try{
                    if(!queue.force(command,timeout,unit)){
                        submittedCount.decrementAndGet();
                    throw new RejectedExecutionException("Queue capacity is full.");
                    }
                }catch(InterruptedException x){
                    submittedCount.decrementAndGet();
                    Thread.interrupted();
                    throw new RejectedExecutionException(x);
                }
                }else{
                    submittedCount.decrementAndGet();
                    throw rx;
                }
            }
        }

TaskQueue.java

public boolean force(Runnable o, long timeout, TimeUnit unit) throws InterruptedException {
 if ( parent.isShutdown() ) 
 throw new RejectedExecutionException(
 "Executor not running, can't force a command into the queue"
 );
 return super.offer(o,timeout,unit); //forces the item onto the queue, to be used if the task 
is rejected
}

Executor 线程配置

配置项	默认值	说明
threadPriority	5	线程优先级
daemon	true	是否守护线程
minSpareThreads	25	核心线程数，即 corePoolSize
maxThreads	200	最大线程数，即 maximumPoolSize
maxIdleTime	60000	线程生存时间，单位是毫秒，默认值即 1 分钟
maxQueueSize	Integer.MAX_VALUE	队列长度
prestartminSpareThreads	false	核心线程是否在服务器启动时启动

Fork/Join

概念

Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 cpu 密集型运算

所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解

Fork/Join 在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率

Fork/Join 默认会创建与 cpu 核心数大小相同的线程池

使用

提交给 Fork/Join 线程池的任务需要继承 RecursiveTask（有返回值）或 RecursiveAction（没有返回值），例如下面定义了一个对 1~n 之间的整数求和的任务

public static void main(String[] args) {
 ForkJoinPool pool = new ForkJoinPool(4);
 System.out.println(pool.invoke(new AddTask1(5)));
}

class AddTask1 extends RecursiveTask<Integer> {
 
 int n;
 
 public AddTask1(int n) {
 this.n = n;
 }
 
 @Override
 public String toString() {
 return "{" + n + '}';
 }
 
 @Override
 protected Integer compute() {
 // 如果 n 已经为 1，可以求得结果了
 if (n == 1) {
 log.debug("join() {}", n);
 return n;
 }

[ForkJoinPool-1-worker-0] - fork() 2 + {1} 
[ForkJoinPool-1-worker-1] - fork() 5 + {4} 
[ForkJoinPool-1-worker-0] - join() 1 
[ForkJoinPool-1-worker-0] - join() 2 + {1} = 3 
[ForkJoinPool-1-worker-2] - fork() 4 + {3} 
[ForkJoinPool-1-worker-3] - fork() 3 + {2} 
[ForkJoinPool-1-worker-3] - join() 3 + {2} = 6 
[ForkJoinPool-1-worker-2] - join() 4 + {3} = 10 
[ForkJoinPool-1-worker-1] - join() 5 + {4} = 15 
15
 
 // 将任务进行拆分(fork)
 AddTask1 t1 = new AddTask1(n - 1);
 t1.fork();
 log.debug("fork() {} + {}", n, t1);
 
 // 合并(join)结果
 int result = n + t1.join();
 log.debug("join() {} + {} = {}", n, t1, result);
 return result;
 }
}

用图来表示