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
