线程池

作者: 我可能是个假开发 | 来源:发表于2023-12-05 12:20 被阅读0次

image.png

一、线程池状态

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 拒绝策略

image.png

工作方式

线程池中刚开始没有线程，当一个任务提交给线程池后，线程池会创建一个新线程来执行任务。
当线程数达到 corePoolSize 并没有线程空闲，这时再加入任务，新加的任务会被加入workQueue 队列排队，直到有空闲的线程
如果阻塞队列选择了有界队列，那么任务超过了队列大小时，会创建 maximumPoolSize - corePoolSize 数目的线程来救急（如果是无界队列则不会有救急线程）。
如果线程到达 maximumPoolSize 仍然有新任务这时会执行拒绝策略。拒绝策略 jdk 提供了 4 种实现，其它著名框架也提供了实现
AbortPolicy 让调用者抛出 RejectedExecutionException 异常，这是默认策略
CallerRunsPolicy 让调用者运行任务
DiscardPolicy 放弃本次任务
DiscardOldestPolicy 放弃队列中最早的任务，本任务取而代之
Dubbo 的实现，在抛出 RejectedExecutionException 异常之前会记录日志，并 dump 线程栈信息，方便定位问题
Netty 的实现，是创建一个新线程来执行任务
ActiveMQ 的实现，带超时等待（60s）尝试放入队列，类似我们之前自定义的拒绝策略
PinPoint 的实现，它使用了一个拒绝策略链，会逐一尝试策略链中每种拒绝策略
当高峰过去后，超过corePoolSize 的救急线程如果一段时间没有任务做，需要结束节省资源，这个时间由keepAliveTime 和 unit 来控制。

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

三、JDK Executors

1.newFixedThreadPool

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

特点:

核心线程数 == 最大线程数（没有救急线程被创建），因此也无需超时时间
阻塞队列是无界的，可以放任意数量的任务

适用于任务量已知，相对耗时的任务

2.newCachedThreadPool

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

特点:

核心线程数是 0，最大线程数是 Integer.MAX_VALUE，救急线程的空闲生存时间是 60s
全部都是救急线程（60s 后可以回收）
救急线程可以无限创建
队列采用了 SynchronousQueue 实现，特点是没有容量，没有线程来取就放不进去

package juc.pool;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.SynchronousQueue;


@Slf4j
public class TestSynchronousQueue {
    public static void main(String[] args) throws InterruptedException {
        SynchronousQueue<Integer> integers = new SynchronousQueue<>();
        new Thread(() -> {
            try {
                log.debug("putting {} ", 1);
                integers.put(1);
                log.debug("{} put ", 1);

                log.debug("putting {} ", 2);
                integers.put(2);
                log.debug("{} put ", 2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t1").start();

        Thread.sleep(1000);

        new Thread(() -> {
            try {
                log.debug("taking {}", 1);
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t2").start();

        Thread.sleep(1000);

        new Thread(() -> {
            try {
                log.debug("taking {}", 2);
                integers.take();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        },"t3").start();
    }
}

16:50:42.880 [t1] DEBUG juc.pool.TestSynchronousQueue - putting 1 
16:50:43.885 [t2] DEBUG juc.pool.TestSynchronousQueue - taking 1
16:50:43.886 [t1] DEBUG juc.pool.TestSynchronousQueue - 1 put 
16:50:43.886 [t1] DEBUG juc.pool.TestSynchronousQueue - putting 2 
16:50:44.886 [t3] DEBUG juc.pool.TestSynchronousQueue - taking 2
16:50:44.886 [t1] DEBUG juc.pool.TestSynchronousQueue - 2 put

线程数会根据任务量不断增长，没有上限，当任务执行完毕，空闲 1分钟后释放线程。适合任务数比较密集，但每个任务执行时间较短的情况

3.newSingleThreadExecutor

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

使用场景：希望多个任务排队执行。线程数固定为 1，任务数多于 1 时，会放入无界队列排队。任务执行完毕，唯一的线程不会被释放。

对比：

对比自己创建一个单线程串行执行任务，如果任务执行失败而终止那么没有任何补救措施，而线程池还会新建一个线程，保证池的正常工作
Executors.newSingleThreadExecutor() 线程个数始终为1：不能修改，
FinalizableDelegatedExecutorService 应用的是装饰模式，只对外暴露了 ExecutorService 接口，因此不能调用 ThreadPoolExecutor 中特有的方法
对比Executors.newFixedThreadPool(1) 初始时为1：以后还可以修改，对外暴露的是 ThreadPoolExecutor 对象，可以强转后调用 setCorePoolSize 等方法进行修改

四、提交任务

// 执行任务
void execute(Runnable command);

// 提交任务 task，用返回值 Future 获得任务执行结果
<T> Future<T> submit(Callable<T> task);

// 提交 tasks 中所有任务
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks) throws InterruptedException;

// 提交 tasks 中所有任务，带超时时间
<T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException;

// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消
<T> T invokeAny(Collection<? extends Callable<T>> tasks) throws InterruptedException, ExecutionException;

// 提交 tasks 中所有任务，哪个任务先成功执行完毕，返回此任务执行结果，其它任务取消，带超时时间
<T> T invokeAny(Collection<? extends Callable<T>> tasks,long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException;

submit():

@Slf4j
public class SubmitTest {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        Future<String> future = pool.submit(() -> {
            log.debug("running");
            Thread.sleep(3000);
            return "ok";
        });
        log.debug("result:{}",future.get());
    }
}

17:08:07.862 [pool-1-thread-1] DEBUG juc.pool.SubmitTest - running
17:08:10.869 [main] DEBUG juc.pool.SubmitTest - result:ok

invokeAll():

@Slf4j
public class SubmitTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        List<Future<String>> futures = pool.invokeAll(Arrays.asList(
                () -> {
                    log.debug("begin");
                    Thread.sleep(1000);
                    return "1";
                },
                () -> {
                    log.debug("begin");
                    Thread.sleep(500);
                    return "2";
                },
                () -> {
                    log.debug("begin");
                    Thread.sleep(2000);
                    return "3";
                }
        ));

        futures.forEach( f ->  {
            try {
                log.debug("{}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
}

19:09:12.570 [pool-1-thread-1] DEBUG juc.pool.SubmitTest - begin
19:09:12.570 [pool-1-thread-2] DEBUG juc.pool.SubmitTest - begin
19:09:13.077 [pool-1-thread-2] DEBUG juc.pool.SubmitTest - begin
19:09:15.081 [main] DEBUG juc.pool.SubmitTest - 1
19:09:15.085 [main] DEBUG juc.pool.SubmitTest - 2
19:09:15.085 [main] DEBUG juc.pool.SubmitTest - 3

invokeAny():

@Slf4j
public class SubmitTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);
        // 谁先执行完返回谁
        String result = pool.invokeAny(Arrays.asList(
                () -> {
                    log.debug("begin");
                    Thread.sleep(1000);
                    log.debug("end");
                    return "1";
                },
                () -> {
                    log.debug("begin");
                    Thread.sleep(500);
                    log.debug("end");
                    return "2";
                },
                () -> {
                    log.debug("begin");
                    Thread.sleep(2000);
                    log.debug("end");
                    return "3";
                }
        ));
        log.debug("result:{}",result);
}

19:20:03.968 [pool-1-thread-1] DEBUG juc.pool.SubmitTest - begin
19:20:03.968 [pool-1-thread-2] DEBUG juc.pool.SubmitTest - begin
19:20:04.474 [pool-1-thread-2] DEBUG juc.pool.SubmitTest - end
19:20:04.474 [pool-1-thread-2] DEBUG juc.pool.SubmitTest - begin
19:20:04.474 [main] DEBUG juc.pool.SubmitTest - result:2

五、关闭线程池

1.shutdown()

/*
线程池状态变为 SHUTD
- 不会接收新任务
- 已提交任务会执行，阻塞队列中的会执行
- 此方法不会阻塞调用
*/
public void shutdown() {
    final ReentrantLock mainLock = this.mainLock;
    mainLock.lock();
    try {
        checkShutdownAccess();
        //修改线程池状态
        advanceRunState(SHUTDOWN);
        // 仅会打断空闲线程
        interruptIdleWorkers();
        // 扩展点 ScheduledThreadPoolExecutor
        onShutdown(); // hook for ScheduledThreadPoolExecutor
    } finally {
        mainLock.unlock();
    }
    // 尝试终结(没有运行的线程可以立刻终结，如果还有运行的线程也不会等)
    tryTerminate();
}

@Slf4j
public class ShutDownTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        Future<Integer> result1 = pool.submit(() -> {
            log.debug("task 1 running...");
            Thread.sleep(1000);
            log.debug("task 1 finish...");
            return 1;
        });

        Future<Integer> result2 = pool.submit(() -> {
            log.debug("task 2 running...");
            Thread.sleep(1000);
            log.debug("task 2 finish...");
            return 2;
        });

        Future<Integer> result3 = pool.submit(() -> {
            log.debug("task 3 running...");
            Thread.sleep(1000);
            log.debug("task 3 finish...");
            return 3;
        });

        log.debug("shutdown");
        pool.shutdown();
        log.debug("after");
    }
}

20:07:39.871 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 1 running...
20:07:39.871 [main] DEBUG juc.pool.ShutDownTest - shutdown
20:07:39.871 [pool-1-thread-2] DEBUG juc.pool.ShutDownTest - task 2 running...
20:07:39.874 [main] DEBUG juc.pool.ShutDownTest - after
20:07:40.877 [pool-1-thread-2] DEBUG juc.pool.ShutDownTest - task 2 finish...
20:07:40.877 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 1 finish...
20:07:40.878 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 3 running...
20:07:41.878 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 3 finish...

@Slf4j
public class ShutDownTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        Future<Integer> result1 = pool.submit(() -> {
            log.debug("task 1 running...");
            Thread.sleep(1000);
            log.debug("task 1 finish...");
            return 1;
        });

        Future<Integer> result2 = pool.submit(() -> {
            log.debug("task 2 running...");
            Thread.sleep(1000);
            log.debug("task 2 finish...");
            return 2;
        });

        Future<Integer> result3 = pool.submit(() -> {
            log.debug("task 3 running...");
            Thread.sleep(1000);
            log.debug("task 3 finish...");
            return 3;
        });

        log.debug("shutdown");
        pool.shutdown();

        //在shutdown之后还执行会报错
        Future<Integer> result4 = pool.submit(() -> {
            log.debug("task 4 running...");
            Thread.sleep(1000);
            log.debug("task 4 finish...");
            return 3;
        });
    }
}

20:06:05.079 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 1 running...
20:06:05.079 [main] DEBUG juc.pool.ShutDownTest - shutdown
20:06:05.079 [pool-1-thread-2] DEBUG juc.pool.ShutDownTest - task 2 running...
Exception in thread "main" java.util.concurrent.RejectedExecutionException: Task java.util.concurrent.FutureTask@53e25b76 rejected from java.util.concurrent.ThreadPoolExecutor@73a8dfcc[Shutting down, pool size = 2, active threads = 2, queued tasks = 1, completed tasks = 0]
    at java.util.concurrent.ThreadPoolExecutor$AbortPolicy.rejectedExecution(ThreadPoolExecutor.java:2063)
    at java.util.concurrent.ThreadPoolExecutor.reject(ThreadPoolExecutor.java:830)
    at java.util.concurrent.ThreadPoolExecutor.execute(ThreadPoolExecutor.java:1379)
    at java.util.concurrent.AbstractExecutorService.submit(AbstractExecutorService.java:134)
    at juc.pool.ShutDownTest.main(ShutDownTest.java:42)
20:06:06.085 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 1 finish...
20:06:06.086 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 3 running...
20:06:06.086 [pool-1-thread-2] DEBUG juc.pool.ShutDownTest - task 2 finish...
20:06:07.089 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 3 finish...

2.shutdownNow()

/*
线程池状态变为 STOP
- 不会接收新任务
- 会将队列中的任务返回
- 并用 interrupt 的方式中断正在执行的任务
*/
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;
}

@Slf4j
public class ShutDownTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        Future<Integer> result1 = pool.submit(() -> {
            log.debug("task 1 running...");
            Thread.sleep(1000);
            log.debug("task 1 finish...");
            return 1;
        });

        Future<Integer> result2 = pool.submit(() -> {
            log.debug("task 2 running...");
            Thread.sleep(1000);
            log.debug("task 2 finish...");
            return 2;
        });

        Future<Integer> result3 = pool.submit(() -> {
            log.debug("task 3 running...");
            Thread.sleep(1000);
            log.debug("task 3 finish...");
            return 3;
        });

        log.debug("shutdown");
        //正在执行的任务会被打断，阻塞队列中的任务会当作返回值返回
        List<Runnable> runnables = pool.shutdownNow();

        log.debug("after:{}",runnables);
    }
}

20:11:13.164 [pool-1-thread-1] DEBUG juc.pool.ShutDownTest - task 1 running...
20:11:13.164 [main] DEBUG juc.pool.ShutDownTest - shutdown
20:11:13.164 [pool-1-thread-2] DEBUG juc.pool.ShutDownTest - task 2 running...
20:11:13.166 [main] DEBUG juc.pool.ShutDownTest - after:[java.util.concurrent.FutureTask@73a8dfcc]

3.其它方法

// 不在 RUNNING 状态的线程池，此方法就返回 true
boolean isShutdown();

// 线程池状态是否是 TERMINATED
boolean isTerminated();

// 调用 shutdown 后，由于调用线程并不会等待所有任务运行结束，因此如果它想在线程池 TERMINATED 后做些事情，可以利用此方法等待
boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException;

七、饥饿

工作线程:
让有限的工作线程(Worker Thread)来轮流异步处理无限多的任务。也可以将其归类为分工模式，它的典型实现就是线程池，也体现了经典设计模式中的享元模式。
不同任务类型应该使用不同的线程池，这样能够避免饥饿，并能提升效率。
固定大小线程池会有饥饿现象：

两个工人是同一个线程池中的两个线程
他们要做的事情是: 为客人点餐和到后厨做菜，这是两个阶段的工作
客人点餐: 必须先点完餐，等菜做好，上菜，在此期间处理点餐的工人必须等待
后厨做菜
工人A 处理了点餐任务，接下来它要等着工人B 把菜做好，然后上菜
但现在同时来了两个客人，这个时候工人A 和工人B 都去处理点餐了，这时没人做饭了，饥饿

@Slf4j
public class HungerTest {

    static final List<String> MENU = Arrays.asList("红烧肉", "豆角肉末", "西红柿炒蛋", "三杯鸡");
    static Random RANDOM = new Random();
    static String cooking() {
        return MENU.get(RANDOM.nextInt(MENU.size()));
    }
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2);

        pool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = pool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
        pool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = pool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}

20:35:03.025 [pool-1-thread-1] DEBUG c.TestDeadLock - 处理点餐...
20:35:03.025 [pool-1-thread-2] DEBUG c.TestDeadLock - 处理点餐...

解决：不同任务类型使用不同的线程池

@Slf4j
public class HungerTest {

    static final List<String> MENU = Arrays.asList("红烧肉", "豆角肉末", "西红柿炒蛋", "三杯鸡");
    static Random RANDOM = new Random();
    static String cooking() {
        return MENU.get(RANDOM.nextInt(MENU.size()));
    }
    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(1);
        ExecutorService waiterPool = Executors.newFixedThreadPool(1);
        ExecutorService cookPool = Executors.newFixedThreadPool(1);

        waiterPool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
        waiterPool.execute(() -> {
            log.debug("处理点餐...");
            Future<String> f = cookPool.submit(() -> {
                log.debug("做菜");
                return cooking();
            });
            try {
                log.debug("上菜: {}", f.get());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }
}

20:38:24.669 [pool-2-thread-1] DEBUG c.TestDeadLock - 处理点餐...
20:38:24.672 [pool-3-thread-1] DEBUG c.TestDeadLock - 做菜
20:38:24.672 [pool-2-thread-1] DEBUG c.TestDeadLock - 上菜: 红烧肉
20:38:24.673 [pool-2-thread-1] DEBUG c.TestDeadLock - 处理点餐...
20:38:24.674 [pool-3-thread-1] DEBUG c.TestDeadLock - 做菜
20:38:24.674 [pool-2-thread-1] DEBUG c.TestDeadLock - 上菜: 红烧肉

八、线程池创建多少线程合适

过小会导致程序不能充分地利用系统资源、容易导致饥饿
过大会导致更多的线程上下文切换，占用更多内存

1.CPU 密集型运算

通常采用 cpu 核数 + 1能够实现最优的 CPU用率，+1 是保证当线程于页缺失故障(操作系统)或其它原因导致暂停时，额外的这个线程就能顶上去，保证 CPU 时钟周期不被浪费

2.I/0 密集型运算

CPU不总是外干繁忙状态，例如，当执行业务计算时，这时候会使用CPU资源，但当执行I/O操作时，远程 RPC 调用时，包括进行数据库操作时，这时候 CPU 就闲下来了，可以利用多线程提高它的利用率。
公式：线程数 = 核数 * 期望 CPU 利用率 * 总时间(CPU计算时间+等待时间) / CPU 计算时间
例如4核CPU 计算时间是 50%，其它等待时间是 50%，期望 cpu 被 100% 利用，套用公式：4100%100% / 50% = 8

例如4核CPU 计算时间是 10%，其它等待时间是 90%，期望 cpu 被 100% 利用，套用公式：4100%100% / 10% = 40

九、任务调度线程池

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

@Slf4j
public class ScheduleTest {

    public static void main(String[] args) throws ExecutionException, InterruptedException {

//        method1();

        ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
        pool.schedule(() -> {
            try {
                log.debug("task1");
//                int i = 1 / 0;
            } catch (Exception e) {
                log.error("error:", e);
            }
        }, 1, TimeUnit.SECONDS);



//        ExecutorService pool = Executors.newFixedThreadPool(1);
//        pool.submit(() -> {
//            try {
//                log.debug("task1");
//                int i = 1 / 0;
//            } catch (Exception e) {
//                log.error("error:", e);
//            }
//        });
    }


    private static void method3() {
        ScheduledExecutorService pool = Executors.newScheduledThreadPool(1);
        log.debug("start...");
        pool.scheduleAtFixedRate(() -> {
            log.debug("running...");
        }, 1, 1, TimeUnit.SECONDS);
    }

    /**
     * 任务1的延迟不会影响到任务2（要保证线程池至少2个线程）
     * 任务1的异常不会影响到任务2
     * @param pool
     */
    private static void method2(ScheduledExecutorService pool) {
        pool.schedule(() -> {
            log.debug("task1");
            int i = 1 / 0;
        }, 1, TimeUnit.SECONDS);

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


    /**
     * task 1的延迟导致了task 2也延迟了
     * task 1发生异常导致task 2无法执行
     */
    private static void method1() {
        Timer timer = new Timer();
        TimerTask task1 = new TimerTask() {
            @Override
            public void run() {
                log.debug("task 1");
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        };
        TimerTask task2 = new TimerTask() {
            @Override
            public void run() {
                log.debug("task 2");
            }
        };
        log.debug("start...");
        timer.schedule(task1, 1000);
        timer.schedule(task2, 1000);
    }
}

十、Tomcat 线程池

image.png

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

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同：
如果总线程数达到 maximumPoolSize，这时不会立刻抛RejectedExecutionException 异常，而是再次尝试将任务放入队列，如果还失败，才抛出 RejectedExecutionException 异常。

Connector 配置：

配置项	默认值	说明
acceptorThreadCount	1	acceptor 线程数量
pollerThreadCount	1	poller 线程数量
minSpareThreads	10	核心线程数，即 corePoolSize
maxThreads	200	最大线程数，即 maximumPoolSize
executor	-	Executor 名称，用来引用下面的 Executor

Executor 线程配置：

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

image.png

线程池

一、线程池状态

二、构造方法

三、JDK Executors

1.newFixedThreadPool

2.newCachedThreadPool

3.newSingleThreadExecutor

四、提交任务

五、关闭线程池

1.shutdown()

2.shutdownNow()

3.其它方法

七、饥饿

八、线程池创建多少线程合适

1.CPU 密集型运算

2.I/0 密集型运算

九、任务调度线程池

十、Tomcat 线程池

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

程序员

@IT·互联网