- 👏作者简介：大家好，我是爱敲代码的小黄，独角兽企业的Java开发工程师，CSDN博客专家，阿里云专家博主
- 📕系列专栏：Java设计模式、Spring源码系列、Netty源码系列、Kafka源码系列、JUC源码系列、duubo源码系列
- 🔥如果感觉博主的文章还不错的话，请👍三连支持👍一下博主哦
- 🍂博主正在努力完成2023计划中：以梦为马，扬帆起航，2023追梦人
- 📝联系方式：hls1793929520，加我进群，大家一起学习，一起进步，一起对抗互联网寒冬👀
ScheduledExecutorService
一、背景
大家好呀，上周我们公司由于定时线程池使用不当出了一个故障，几千万的单子可能没了
给兄弟们分享分享这个坑，希望兄弟们以后别踩！
业务中大量的使用定时线程池（ScheduledExecutorService）执行任务，有时候会忽略掉 Try/Catch 的异常判断
当任务执行报错时，会导致整个定时线程池挂掉，影响业务的正常需求
二、问题
我们来模仿一个生产的例子：
-
合作方修改频率低且合作方允许最终一致性
-
我们有一个定时任务每隔
60秒去MySQL拉取全量的合作方数据放至 合作方缓存（本地缓存） 中 -
当客户请求时，我们去缓存中拿取合作方即可
这样的生产例子应该存在于绝大数公司，代码如下：
public class Demo {
// 创建定时线程池
private static ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
private List partnerCache = new ArrayList();
@PostConstruct
public void init() {
scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
// 启动时每隔60秒执行一次数据库的刷新
// 将数据缓存至本地
loadPartner();
}
}, 3, 60, TimeUnit.SECONDS);
}
public void loadPartner() {
// 查询数据库当前最新合作方数据
List partnerList = queryPartners();
// 合作方数据放至缓存
partnerCache.clear();
partnerCache.addAll(partnerList);
}
public List queryPartners() {
// 数据库挂了！
throw new RuntimeException();
}
}
运行上述样例，我们会发现程序不停止，输出一遍 Load start!，一直在运行，但后续不输出 Load start!
这个时候我们可以确认：异常确实导致当前任务不再执行
1、为什么任务报错会影响定时线程池？
2、定时线程池是真的挂掉了嘛?
3、定时线程池内部是如何执行的？
跟着这三个问题，我们一起来看一看 ScheduledExecutorService 的原理介绍
三、原理剖析
对于 ScheduledExecutorService 来说，本质上是 延时队列 + 线程池
1、延时队列介绍
DelayQueue 是一个无界的 BlockingQueue，用于放置实现了Delayed接口的对象，只能在到期时才能从队列中取走。
这种队列是有序的，即队头对象的延迟到期时间最长。
我们看一下延时队列里对象的属性：
class MyDelayedTask implements Delayed{
// 当前任务创建时间
private long start = System.currentTimeMillis();
// 延时时间
private long time ;
// 初始化
public MyDelayedTask(long time) {
this.time = time;
}
/**
* 需要实现的接口，获得延迟时间（用过期时间-当前时间）
*/
@Override
public long getDelay(TimeUnit unit) {
return unit.convert((start+time) - System.currentTimeMillis(),TimeUnit.MILLISECONDS);
}
/**
* 用于延迟队列内部比较排序（当前时间的延迟时间 - 比较对象的延迟时间）
*/
@Override
public int compareTo(Delayed o) {
MyDelayedTask o1 = (MyDelayedTask) o;
return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
}
}
所以，延时队列的实现原理也很简单：
- 生产端：投递消息时增加时间戳（当前时间+延时时间）
- 消费端：用当前时间与时间戳进行比较，若小于则消费，反之则循环等待
2、线程池的原理介绍
- 当前的线程池个数低于核心线程数，直接添加核心线程即可
- 当前的线程池个数大于核心线程数，将任务添加至阻塞队列中
- 如果添加阻塞队列失败，则需要添加非核心线程数处理任务
- 如果添加非核心线程数失败（满了），执行拒绝策略
3、定时线程的原理
我们从定时线程池的创建看：scheduledExecutorService.scheduleAtFixedRate(myTask, 3L, 1L, TimeUnit.SECONDS);
public ScheduledFuture scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
// 初始化我们的任务
// triggerTime：延时的实现
ScheduledFutureTask sft = new ScheduledFutureTask(command,null,triggerTime(initialDelay, unit),unit.toNanos(period));
RunnableScheduledFuture t = decorateTask(command, sft);
sft.outerTask = t;
delayedExecute(t);
return t;
}
private void delayedExecute(RunnableScheduledFuture task) {
// 将当前任务丢进延时队列
super.getQueue().add(task);
// 创建核心线程并启动
ensurePrestart();
}
// 时间轮算法
private long triggerTime(long delay, TimeUnit unit) {
return now() + delay;
}
从这里我们可以得到结论：定时线程池通过延时队列来达到定时的目的
有一个问题：我们仅仅向 Queue 里面放了一个任务，他是怎么保证执行多次的呢？
带着这个问题，我们看一下他拉取任务启动的代码：
for (;;) {
// 从延时队列中获取任务
Runnable r = workQueue.take();
}
public RunnableScheduledFuture take(){
for (;;) {
// 获取队列第一个任务
RunnableScheduledFuture first = queue[0];
// 【重点】如果当前队列任务为空，则等待
if (first == null){
available.await();
}
// 获取当前任务的时间
long delay = first.getDelay(NANOSECONDS);
if (delay
