java并发基础-线程池

2022-06-08 本文已影响0人东南枝下

线程池主要解决两个问题：

当执行大量异步任务时线程池能够提供较好的性能。在不使用线程池时，每当需要执行异步任务时直接new一个线程来运行，而线程的创建和销毁是需要开销的。线程池里面的线程是可复用的，不需要每次执行异步任务时都重新创建和销毁线程。

线程池提供了一种资源限制和管理的手段，比如可以限制线程的个数，动态新增线程等

Executors

JDK并发包中的Executors 提供了很多创建线程池的方法，但实际工作中不建议使用，手动创建线程更可控

ThreadPoolExecutor

查看Executors.newFixedThreadPool,可以看出内部调用的就是new ThreadPoolExecutor的构造函数。
返回值是ExecutorService，ThreadPoolExecutor的父类也实现了ExecutorService接口。

ThreadPoolExecutor的构造函数有以下几个关键参数:

corePoolSize：指定了线程池中的线程数量。
maximumPoolSize：指定了线程池中的最大线程数量。
keepAliveTime：存活时间，当线程池中线程数大于核心线程数corePoolSize，并且是闲置状态，这些闲置线程的存活时间。
unit：keepAliveTime的单位。
workQueue：任务队列，被提交但尚未被执行的任务。
- 是一个阻塞队列BlockingQueue接口的对象，仅用于存放Runnable对象。比如基于数组的有界ArrayBlockingQueue、基于链表的无界LinkedBlockingQueue、最多只有一个元素的同步队列SynchronousQueue及优先级队列PriorityBlockingQueue等。
threadFactory：线程工厂，用于创建线程，一般用默认的即可。
handler：拒绝策略。当任务太多来不及处理时，如何拒绝任务。
JDK内置拒绝策略:
1. AbortPolicy策略：该策略会直接抛出异常，阻止系统正常工作。
2. CallerRunsPolicy策略：只要线程池未关闭，该策略直接在调用者线程中，运行当前被丢弃的任务。
3. DiscardOldestPolicy策略：该策略将丢弃最老的一个请求，也就是即将被执行的一个任务，并尝试再次提交当前任务。
4. DiscardPolicy策略：该策略默默地丢弃无法处理的任务，不予任何处理。
5. 实现RejectedExecutionHandler接口，自定义拒绝策略。

demo

package com.jenson.pool;

import java.io.IOException;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class ThreadPoolExecutorTest {
    public static void main(String[] args) throws IOException {
        // 核心线程数
        int corePoolSize = 2;
        // 线程池最大线程数
        int maximumPoolSize = 5;
        // 存活时间，当线程池中线程数大于核心线程数，并且是闲置状态，这些闲置线程的存活时间
        long keepAliveTime = 10;
        // 存活时间单位
        TimeUnit unit = TimeUnit.SECONDS;
        // 用于保存等待执行(提交但未执行)的任务的阻塞队列，
        // 比如基于数组的有界ArrayBlockingQueue、
        // 基于链表的无界LinkedBlockingQueue、
        // 最多只有一个元素的同步队列SynchronousQueue及优先级队列PriorityBlockingQueue等。
        // 有界队列仅当在任务队列装满时，才可能将线程数提升到corePoolSize以上,若大于maximumPoolSize，则会执行拒绝策略
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(2);
        // 无界队列线程数不会超过corePoolSize，线程会一直加入队列不会触发拒绝策略，直至系统资源耗尽
//        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();
        // 创建线程的工厂
        ThreadFactory threadFactory = new NameTreadFactory();
        // 拒绝策略，当队列满并且线程个数达到maximunPoolSize后采取的策略，
        // 比如AbortPolicy（抛出异常）、CallerRunsPolicy（使用调用者所在线程来运行任务）、
        // DiscardOldestPolicy（调用poll丢弃一个任务，执行当前任务）及DiscardPolicy（默默丢弃，不抛出异常）
        RejectedExecutionHandler handler = new MyIgnorePolicy();
        ThreadPoolExecutor pool = new ThreadPoolExecutor(corePoolSize,
                maximumPoolSize,
                keepAliveTime,
                unit,
                workQueue,
                threadFactory,
                handler);
        // 预启动所有核心线程
        pool.prestartAllCoreThreads();

        for (int i = 1; i <= 20; i++) {
            MyTask task = new MyTask(String.valueOf(i));
            // execute只能提交Runnable类型的任务，无返回值。会直接抛出任务执行时异常
            pool.execute(task);
            // submit既能提交Runable类型的任务，返回值为null，也能提交Callable类型的任务，返回值为Future。
            // Future的get方法将任务执行时的异常重新抛出。
//            pool.submit(task);
        }

        System.in.read(); //阻塞主线程
    }


    static class NameTreadFactory implements ThreadFactory {
        private final AtomicInteger mThreadNum = new AtomicInteger(1);

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "my-thread-" + mThreadNum.getAndIncrement());
            System.out.println(t.getName() + " has been created");
            return t;
        }
    }

    public static class MyIgnorePolicy implements RejectedExecutionHandler {

        @Override
        public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
            doLog(r, executor);
        }

        private void doLog(Runnable r, ThreadPoolExecutor e) {
            // 可做日志记录等
            System.err.println(r.toString() + " rejected !completedTaskCount: " + e.getCompletedTaskCount());
        }
    }

    static class MyTask implements Runnable {
        private String name;

        public MyTask(String name) {
            this.name = name;
        }

        @Override
        public void run() {
            System.out.println(this.toString() + " is running!");
            try {
                //让任务执行慢点
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        public String getName() {
            return name;
        }

        @Override
        public String toString() {
            return "MyTask [name=" + name + "]";
        }
    }
}