使用Java實(shí)現(xiàn)一個(gè)簡(jiǎn)單的定時(shí)器
1. 概要
說到定時(shí)任務(wù)，大家都不會(huì)模式，前幾天在看到 Java 的 Timer 定時(shí)（延時(shí)）任務(wù)，就順著 Timer - ScheduledThreadPoolExecutor - RocketMQ - Netty 這些框架看了下定時(shí)任務(wù)的一些核心邏輯實(shí)現(xiàn)，現(xiàn)在就寫一個(gè)系列的文章來記錄下這些源碼的實(shí)現(xiàn)過程，如果有錯(cuò)誤的地方歡迎指出。
定時(shí)任務(wù)在我們平時(shí)的項(xiàng)目中都會(huì)多多少少有接觸，在很多場(chǎng)景下都非常有用，比如：
- 每天批量處理數(shù)據(jù)、生成報(bào)告等
- 會(huì)議提醒、郵件發(fā)送等
- 訂單支付超時(shí)退款、清理臨時(shí)文件、日志文件…
下面在介紹切入正題之前我們先自己實(shí)現(xiàn)一個(gè)最簡(jiǎn)單的定時(shí)/延時(shí)器
2. 注意問題
要實(shí)現(xiàn)一個(gè)定時(shí)任務(wù)，首先我們要考慮以下幾點(diǎn)問題
- 如何讓任務(wù)按照?qǐng)?zhí)行的時(shí)間排序（什么樣的集合）
- 定時(shí)任務(wù)到期之后如何執(zhí)行
- 如何實(shí)現(xiàn)任務(wù)執(zhí)行多次
帶著這三個(gè)問題，我們來考慮定時(shí)器的實(shí)現(xiàn)
首先是第一個(gè)問題，什么樣的集合能滿足我們的要求，需要滿足按照固定順序自動(dòng)排序調(diào)整任務(wù)，用 JDK 的 PriorityQueue 優(yōu)先隊(duì)列就可以了，只需要設(shè)置按照任務(wù)的執(zhí)行時(shí)間排序，當(dāng)我們把任務(wù)添加到隊(duì)列里面，隊(duì)列就會(huì)按照任務(wù)的執(zhí)行時(shí)間自動(dòng)排序，這部分代碼就不用我們考慮了
第二個(gè)問題，我們可以額外啟動(dòng)一個(gè)任務(wù)處理線程專門去循環(huán)隊(duì)列處理里面的到期任務(wù)，如果任務(wù)沒到期就阻塞等待，如果到期了就喚醒執(zhí)行任務(wù)
第三個(gè)問題，添加任務(wù)的時(shí)候可以設(shè)置一個(gè) Count 表示任務(wù)需要執(zhí)行的次數(shù)，添加任務(wù)的時(shí)候就可以 for 循環(huán)去遍歷添加
3. 實(shí)現(xiàn)
3.1 任務(wù)定義
首先定義一個(gè)任務(wù) bean，里面包括任務(wù)的執(zhí)行時(shí)間和執(zhí)行的任務(wù)：
class Task {
// 執(zhí)行時(shí)間
private long executeTime;
// 任務(wù)
Runnable runnable;
public Task(long executeTime, Runnable runnable) {
this.executeTime = executeTime;
this.runnable = runnable;
}
public long getExecuteTime() {
return executeTime;
}
public void setExecuteTime(long executeTime) {
this.executeTime = executeTime;
}
public Runnable getRunnable() {
return runnable;
}
public void setRunnable(Runnable runnable) {
this.runnable = runnable;
}
}
3.2 添加任務(wù)的方法
public boolean add(long addTime, Runnable runnable, int count) {
long now = System.currentTimeMillis();
synchronized (queue) {
for (int i = 0; i < count; i++) {
// 計(jì)算任務(wù)的執(zhí)行時(shí)間
long newTime = now + addTime * (i + 1);
// 如果小于 now，說明溢出了
if (newTime < now) {
throw new RuntimeException(Thread.currentThread().getName() + ": overflow long limit");
}
Task min = getMin();
if (min == null || newTime < min.executeTime) {
// 加入任務(wù)隊(duì)列
queue.offer(new Task(newTime, runnable));
// 喚醒任務(wù)線程
queue.notifyAll();
continue;
}
// 加入任務(wù)隊(duì)列
queue.offer(new Task(newTime, runnable));
}
}
return true;
}首先計(jì)算任務(wù)的執(zhí)行時(shí)間，判斷一下是否會(huì)溢出，如果溢出就拋出異常
再獲取隊(duì)首任務(wù)，如果添加的新任務(wù)執(zhí)行時(shí)間比隊(duì)首任務(wù)更塊，調(diào)用 queue.notifyAll() 喚醒線程去執(zhí)行任務(wù)
將任務(wù)加入隊(duì)列中
3.3 任務(wù)線程
Thread thread = new Thread(() -> {
System.out.println(Thread.currentThread().getName() + ": 啟動(dòng) execute-job-thread 線程");
while (true) {
try {
synchronized (queue) {
if (queue.isEmpty()) {
this.queue.wait();
}
long now = System.currentTimeMillis();
while (!queue.isEmpty() && now >= queue.peek().getExecuteTime()) {
Task task = queue.poll();
// 線程池執(zhí)行任務(wù)
executor.execute(task.runnable);
}
if (queue.isEmpty()) {
this.queue.wait();
} else {
long div = queue.peek().executeTime - now;
this.queue.wait(div);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}, "execute-job-thread");線程在一個(gè)死循環(huán)里面不斷遍歷隊(duì)列看有沒有任務(wù)值
- 如果沒有任務(wù)，就阻塞等待
- 被喚醒后代表有任務(wù)，這時(shí)候獲取當(dāng)前時(shí)間，獲取所有的過期的任務(wù)依次執(zhí)行
- 執(zhí)行完之后根據(jù)隊(duì)列中是否還有任務(wù)來判斷要阻塞多長(zhǎng)時(shí)間
3.4 整體代碼
PQueueService 代碼如下：
public class PQueueService {
private static final int MAX_COUNT = Integer.MAX_VALUE;
private final PriorityQueue<Task> queue = new PriorityQueue<>((o1, o2) -> {
return Long.compare(o1.executeTime, o2.executeTime);
});
static final ThreadPoolExecutor executor =
new ThreadPoolExecutor(16, 32, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(16));
Thread thread = new Thread(() -> {
System.out.println(Thread.currentThread().getName() + ": 啟動(dòng) execute-job-thread 線程");
while (true) {
try {
synchronized (queue) {
if (queue.isEmpty()) {
this.queue.wait();
}
long now = System.currentTimeMillis();
while (!queue.isEmpty() && now >= queue.peek().getExecuteTime()) {
Task task = queue.poll();
// 線程池執(zhí)行任務(wù)
executor.execute(task.runnable);
}
if (queue.isEmpty()) {
this.queue.wait();
} else {
long div = queue.peek().executeTime - now;
this.queue.wait(div);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}, "execute-job-thread");
public void init() {
thread.start();
}
public Task getMin() {
if (queue.isEmpty()) {
return null;
}
synchronized (queue) {
return queue.peek();
}
}
public Task pollMin() {
if (queue.isEmpty()) {
return null;
}
synchronized (queue) {
return queue.poll();
}
}
public boolean add(long addTime, Runnable runnable, int count) {
long now = System.currentTimeMillis();
synchronized (queue) {
for (int i = 0; i < count; i++) {
long newTime = now + addTime * (i + 1);
if (newTime < now) {
throw new RuntimeException(Thread.currentThread().getName() + ": overflow long limit");
}
Task min = getMin();
if (min == null || newTime < min.executeTime) {
// 加入任務(wù)隊(duì)列
queue.offer(new Task(newTime, runnable));
// 喚醒任務(wù)線程
queue.notifyAll();
continue;
}
// 加入任務(wù)隊(duì)列
queue.offer(new Task(newTime, runnable));
}
}
return true;
}
class Task {
private long executeTime;
Runnable runnable;
public Task(long executeTime, Runnable runnable) {
this.executeTime = executeTime;
this.runnable = runnable;
}
public long getExecuteTime() {
return executeTime;
}
public void setExecuteTime(long executeTime) {
this.executeTime = executeTime;
}
public Runnable getRunnable() {
return runnable;
}
public void setRunnable(Runnable runnable) {
this.runnable = runnable;
}
}
}
Timer 類代碼如下：
public class Timer {
public static final Logger logger = LoggerFactory.getLogger(Timer.class);
PQueueService pQueueService;
public Timer() {
pQueueService = new PQueueService();
pQueueService.init();
}
/**
* 執(zhí)行任務(wù)
*/
public void fixExecuteOne(int seconds, TimeUnit timeUnit, Runnable runnable, int count) {
if (timeUnit.ordinal() != TimeUnit.SECONDS.ordinal()) {
throw new RuntimeException("unknown unit");
}
long time = seconds * 1000L;
pQueueService.add(time, runnable, count);
}
public static final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
public static void main(String[] args) {
Timer timer = new Timer();
// 1.添加一個(gè)10s后執(zhí)行的任務(wù)
timer.fixExecuteOne(10, TimeUnit.SECONDS, () -> {
System.out.println("10s任務(wù) ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
// 2.添加一個(gè)3s后執(zhí)行的任務(wù)
timer.fixExecuteOne(3, TimeUnit.SECONDS, () -> {
System.out.println("3s任務(wù) ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
// 3.添加一個(gè)20s后執(zhí)行的任務(wù)
timer.fixExecuteOne(20, TimeUnit.SECONDS, () -> {
System.out.println("20s任務(wù) ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
// 4.添加一個(gè)1s后執(zhí)行的任務(wù)
timer.fixExecuteOne(1, TimeUnit.SECONDS, () -> {
System.out.println("1s任務(wù)1 ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
// 5.添加一個(gè)1s后執(zhí)行的任務(wù)
timer.fixExecuteOne(1, TimeUnit.SECONDS, () -> {
System.out.println("1s任務(wù)2 ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
// 6.添加一個(gè)1s后執(zhí)行的任務(wù)
timer.fixExecuteOne(1, TimeUnit.SECONDS, () -> {
System.out.println("1s任務(wù)3 ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 2);
try {
Thread.sleep(30000);
// 7.添加一個(gè)1s后執(zhí)行的任務(wù)
timer.fixExecuteOne(1, TimeUnit.SECONDS, () -> {
System.out.println("1s任務(wù)3 ===> " + Thread.currentThread().getName() + ": " + "執(zhí)行了!!!!!, 當(dāng)前時(shí)間：" + simpleDateFormat.format(new Date()));
}, 1);
Thread.sleep(30000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
執(zhí)行結(jié)果：
execute-job-thread: 啟動(dòng) execute-job-thread 線程
1s任務(wù)2 ===> pool-1-thread-2: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:21
1s任務(wù)1 ===> pool-1-thread-1: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:21
1s任務(wù)3 ===> pool-1-thread-3: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:21
1s任務(wù)1 ===> pool-1-thread-4: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:22
1s任務(wù)2 ===> pool-1-thread-5: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:22
1s任務(wù)3 ===> pool-1-thread-6: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:22
3s任務(wù) ===> pool-1-thread-7: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:23
3s任務(wù) ===> pool-1-thread-8: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:26
10s任務(wù) ===> pool-1-thread-9: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:30
10s任務(wù) ===> pool-1-thread-10: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:40
20s任務(wù) ===> pool-1-thread-11: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:40
30s后添加的1s任務(wù) ===> pool-1-thread-12: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:17:51
20s任務(wù) ===> pool-1-thread-13: 執(zhí)行了!!!!!, 當(dāng)前時(shí)間：2024-11-23 21:18:00
4. 存在問題
上面實(shí)現(xiàn)了一個(gè)簡(jiǎn)單的定時(shí)器，但是這個(gè)定時(shí)器還是存在下面一些問題：
- 功能比較簡(jiǎn)單，沒有其他的 remove 等操作
- 任務(wù)線程拿到任務(wù)之后會(huì)丟到另外一個(gè)線程池中執(zhí)行，如果線程池中任務(wù)比較多，就會(huì)阻塞導(dǎo)致執(zhí)行時(shí)間不準(zhǔn)
- 代碼中對(duì) queue 進(jìn)行了加鎖，但是一個(gè)任務(wù)可以添加到多個(gè) queue，這里沒有做并發(fā)處理
- …
總之，這就是一個(gè)簡(jiǎn)單的 Timer Demo，主要了解下定時(shí)器/延時(shí)器的執(zhí)行過程， 但是其實(shí) JDK 里面的定時(shí)器/延時(shí)器實(shí)現(xiàn)流程都大差不差，下一篇文章，我們就來看看 Timer 的用法和源碼
到此這篇關(guān)于使用Java實(shí)現(xiàn)一個(gè)簡(jiǎn)單的定時(shí)器的文章就介紹到這了,更多相關(guān)Java定時(shí)器內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
相關(guān)文章
這篇文章主要介紹了Spring及Mybatis整合占位符解析失敗問題解決,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-07-07
MyBatis Plus實(shí)現(xiàn)一對(duì)多的查詢場(chǎng)景的三種方法
MyBatis Plus提供了多種簡(jiǎn)便的方式來進(jìn)行一對(duì)多子查詢,本文主要介紹了MyBatis Plus實(shí)現(xiàn)一對(duì)多的查詢場(chǎng)景的三種方法,具有一定的參考價(jià)值,感興趣的可以了解一下2024-07-07
這篇文章主要以圖文結(jié)合的方式詳細(xì)介紹了Java環(huán)境變量配置方法，文中步驟介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2016-04-04
這篇文章主要介紹了Java變量，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2021-08-08
這篇文章主要介紹了SpringBoot的攔截器中依賴注入為null的解決方法，小編覺得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過來看看吧2018-06-06
Java?Stream?的?sorted實(shí)現(xiàn)自定義排序從基礎(chǔ)到高級(jí)技巧
本文將介紹如何使用sorted()方法實(shí)現(xiàn)自定義排序,涵蓋各種常見場(chǎng)景和高級(jí)技巧,通過掌握?sorted()?方法的各種用法,你可以靈活應(yīng)對(duì)各種復(fù)雜的排序需求,編寫出簡(jiǎn)潔、高效且易于維護(hù)的代碼,需要的朋友跟隨小編一起學(xué)習(xí)吧2025-08-08
Java SiteMesh新手學(xué)習(xí)教程代碼案例
這篇文章主要介紹了Java SiteMesh新手學(xué)習(xí)教程代碼案例,文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-10-10
解決try-catch捕獲異常信息后Spring事務(wù)失效的問題
這篇文章主要介紹了解決try-catch捕獲異常信息后Spring事務(wù)失效的問題，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-06-06
IDEA如何切換數(shù)據(jù)庫(kù)版本mysql5或mysql8
本文介紹了如何將IntelliJ IDEA從MySQL5切換到MySQL8的詳細(xì)步驟,包括下載MySQL8、安裝、配置、停止舊服務(wù)、啟動(dòng)新服務(wù)以及更改密碼等2025-01-01
這篇文章主要介紹了Java冒泡排序法和選擇排序法的實(shí)現(xiàn)，文中通過示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-09-09