线程池总结

一：进程、线程、协程的区别

1.进程是操作系统分配和调度的最小单元，进程有独立的资源空间。
2.线程是CPU调度的最小单元，一个进程可以有多个线程
3.协程是轻量级线程，JDK21之后可以直接创建协程进而减少用户态和内存态的切换。

二：java中创建线程的方式有几种

1.创建Runnable对象使用Thread.start运行线程。
2.创建Callable对象使用Exc.submit运行线程。
3.使用线程池创建对象
4.继承Thread

 

 

三：Future和FutureTask和CompletableFuture的区别

1.Future

1.Future表示一个异步任务的结果，它提供了检查任务是否已完成、以及等待任务完成并获取结果等方法

2.get()方法阻塞获取

ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());

Future<?> future = threadPoolExecutor.submit(() -> {
    //代码逻辑执行
});
future.get();
future.cancel();
future.isCancelled();
future.isDone()

 

2.FutureTask

 

1.FutureTask实现了RunnableFuture,而RunnableFuture继承了Runnable和Future，所以FutureTask即可获取结果又可被线程执行。
2.创建FutureTask需要实现Runnable或Callable，使用Executor的submit执行Callable的call方法,使用excute执行执行Runnable的run方法，使用Thread.start启动线程。

3.get()方法阻塞获取。

        //FutureTask执行CallableFutureTask callableTask = new FutureTask(new Callable() {@Overridepublic Object call() throws Exception {return "1234";}});//线程执行new Thread(callableTask).start();try {Object o = callableTask.get();System.out.println(o);} catch (Exception e) {e.printStackTrace();}

//FutureTask执行RunnableFutureTask runnableTask = new FutureTask(new Runnable() {@Overridepublic void run() {System.out.println(123);}}, false);//线程执行new Thread(runnableTask).start();}

 

3.CompletableFuture

1.CompletableFuture是Java 8引入的Future扩展，它提供了更强大的异步编程能力，支持链式调用、组合任务和回调处理。

2.可以通过thenRun() 、thenAccept()、thenApply() 方法将前后任务连接起来，形成前后有依赖的任务链。其中：

• thenRun(Runnable runnable): 对异步任务的结果进行操作，不能传入参，也没有返回值。
• thenAccept(Consumer consumer): 对异步任务的结果进行消费，可以传入参，但没有返回值。
• thenApply(Function function): 对异步任务的结果进行转换，可传入参，返回一个新的CompletableFuture。

3.‌并行聚合‌，同一超时控制

        //带响应结果CompletableFuture<String> supplyAsync = CompletableFuture.supplyAsync(() -> {//执行逻辑return "返回响应结果";}, threadPoolExecutor);//不带响应结果CompletableFuture<Void> runAsync = CompletableFuture.runAsync(() -> {//执行逻辑无需响应
        }, threadPoolExecutor);try {//共同设置超时时间,一个失败全部失败CompletableFuture.allOf(supplyAsync,runAsync).get(500,TimeUnit.MILLISECONDS);//带响应结果String result = supplyAsync.get();//不带响应结果
            runAsync.get();} catch (Exception e) {e.printStackTrace();}

 

 

四：线程池的创建

 

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(5, 5, 60, TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(100), new ThreadPoolExecutor.CallerRunsPolicy());

 

五：到底如何设置线程池配置

1.核心线程的设置

1.确认机器配置2C4G、4C8G
2.确认业务类型是：CPU密集型、IO密集型，CPU密集型是基于内存计算较多，比如数据计算、逻辑判断等等，IO密集型是磁盘IO/网络IO,比如调用数据库、Redis、第三方接口等等，这些都有IO操作和等等时间。
CPU密集型核心线程设置一般都是机器核数±1,因为机器型号、性能存在差异具体多少核心线程性能最好需要压测判断。
IO密集型核心线程理论上有公式，但是由于业务的操作不一样，如访问MYSQL、第三方接口耗时不一样，所以无法具体评估，需要根据压测结果进行判断。线程池的配置可以自定义方法或使用第三方插件控制

 

2.最大线程数的设置

1.核心线程已经可以做到最大利用CPU的性能了，最大线程数一般都是等于核心线程数，如果核心线程已经找到最合理的值，最大线程在追加几个线程可能会影响接口性能。

3.最大存活时间

1.一般都是给60S，没有太多要求

4.队列长度

1.判断业务能接受接口延迟时间是多久
2.判断接口的平均耗时是多少
3.延迟时间除于接口耗时乘核心线程数

有界队列：
ArrayBlockingQueue：基于数组实现需要配合固定容量，当队列满时会触发拒绝策略，适合需要控制任务数量的场景
无界队列：
LinkedBlockingQueue：基于链表实现，最大容量是Integer.MAX_VALUE,但有可能造成OOM，适用处理速度慢但是不能拒绝的任务
LinkedBlockingQueue：也可以指定容量
无缓冲队列：
SynchronousQueue‌：不存储任何元素，每个插入操作必须等待对应的删除操作，生产者和消费者线程必须严格同步，适用于高吞吐常见。
优先级队列：
PriorityBlockingQueue‌：通过二叉树实现，确保每次操作返回队列最高/最低的元素，适用于有优先级和资源控制的常见

 

 

5.拒绝策略

1.根据业务重要程度来评估，如果是打印日志的丢就丢了，如果是核心业务一般都是交给调用线程执行

CallerRunsPolicy:将任务交给调用者
AbortPolicy:拒绝任务并排除异常
DiscardPolicy:丢弃任务
DiscardOldestPolicy:丢弃排队时间最长的任务，尝试把自己加进去

2.如果这4种拒绝策略不能满足还可以自己实现RejectedExecutionHandler

   public static class CallerRunsPolicy implements RejectedExecutionHandler {public CallerRunsPolicy() { }
public void rejectedExecution(Runnable r, ThreadPoolExecutor e) {if (!e.isShutdown()) {r.run();}}}

 

3.线程池的执行原理

 

4.线程池工作线程和普通线程有什么区别

1.线程池工作线程和普通线程本质是没有区别的，如果想用线程只能thread.start()。

2.但是线程池的工作线程封装了一个worker对象，worker对象适配了线程池的逻辑，他实现了Runnable和继承了AQS,为了满足线程池的shutdown和shutdownNow。

//线程执行方法
public void execute(Runnable command) {
    if (command == null)
        throw new NullPointerException();

    int c = ctl.get();
    if (workerCountOf(c) < corePoolSize) {
        if (addWorker(command, true))
            return;
        c = ctl.get();
    }
    if (isRunning(c) && workQueue.offer(command)) {
        int recheck = ctl.get();
        if (! isRunning(recheck) && remove(command))
            reject(command);
        else if (workerCountOf(recheck) == 0)
            addWorker(null, false);
    }
    else if (!addWorker(command, false))
        reject(command);
}

//worker对象
private final class Worker
    extends AbstractQueuedSynchronizer
    implements Runnable
{


    final Thread thread;

    Runnable firstTask;

    volatile long completedTasks;

  
    Worker(Runnable firstTask) {
        setState(-1); // inhibit interrupts until runWorker
        this.firstTask = firstTask;
        this.thread = getThreadFactory().newThread(this);
    }

 

5.如何在线程池执行任务前后追加逻辑

1.自定义类xx继承ThreadPoolExecutor

2.实现ThreadPoolExecutor中的方法

    protected void beforeExecute(Thread t, Runnable r) { }protected void afterExecute(Runnable r, Throwable t) { }