SpringBoot定時任務功能怎麼實現
- WBOY轉載
- 2023-05-10 16:16:131415瀏覽
一背景
專案中需要一個可以動態新增定時定時任務的功能，現在專案中使用的是xxl-job定時任務排程系統，但是經過一番對xxl-job功能的了解，發現xxl-job對專案動態新增定時任務，動態刪除定時任務的支援並不是那麼好，所以需要自己手動實作一個定時任務的功能
二動態定時任務排程
1 技術選擇
Timer 或 ScheduledExecutorService
這兩個都能實現定時任務調度，先看下Timer的定時任務調度
public class MyTimerTask extends TimerTask {
private String name;
public MyTimerTask(String name){
this.name = name;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
@Override
public void run() {
//task
Calendar instance = Calendar.getInstance();
System.out.println(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(instance.getTime()));
}
}
Timer timer = new Timer();
MyTimerTask timerTask = new MyTimerTask("NO.1");
//首次执行,在当前时间的1秒以后,之后每隔两秒钟执行一次
timer.schedule(timerTask,1000L,2000L);在看下ScheduledThreadPoolExecutor的實現
//org.apache.commons.lang3.concurrent.BasicThreadFactory
ScheduledExecutorService executorService = new ScheduledThreadPoolExecutor(1,
new BasicThreadFactory.Builder().namingPattern("example-schedule-pool-%d").daemon(true).build());
executorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
//do something
}
},initialDelay,period, TimeUnit.HOURS);兩個都能實現定時任務，那他們的區別呢，使用阿里p3c會給出建議和區別
多线程并行处理定时任务时，Timer运行多个TimeTask时，只要其中之一没有捕获抛出的异常，其它任务便会自动终止运行，使用ScheduledExecutorService则没有这个问题。
從建議來看，是一定要選擇ScheduledExecutorService了，我們看看原始碼看看為什麼Timer出現問題會終止執行
/**
* The timer thread.
*/
private final TimerThread thread = new TimerThread(queue);
public Timer() {
this("Timer-" + serialNumber());
}
public Timer(String name) {
thread.setName(name);
thread.start();
}新建物件時，我們看到開啟了一個線程，那麼這個線程在做什麼呢？一起看看
class TimerThread extends Thread {
boolean newTasksMayBeScheduled = true;
/**
* 每一件一个任务都是一个quene
*/
private TaskQueue queue;
TimerThread(TaskQueue queue) {
this.queue = queue;
}
public void run() {
try {
mainLoop();
} finally {
// Someone killed this Thread, behave as if Timer cancelled
synchronized(queue) {
newTasksMayBeScheduled = false;
queue.clear(); // 清除所有任务信息
}
}
}
/**
* The main timer loop. (See class comment.)
*/
private void mainLoop() {
while (true) {
try {
TimerTask task;
boolean taskFired;
synchronized(queue) {
// Wait for queue to become non-empty
while (queue.isEmpty() && newTasksMayBeScheduled)
queue.wait();
if (queue.isEmpty())
break; // Queue is empty and will forever remain; die
// Queue nonempty; look at first evt and do the right thing
long currentTime, executionTime;
task = queue.getMin();
synchronized(task.lock) {
if (task.state == TimerTask.CANCELLED) {
queue.removeMin();
continue; // No action required, poll queue again
}
currentTime = System.currentTimeMillis();
executionTime = task.nextExecutionTime;
if (taskFired = (executionTime<=currentTime)) {
if (task.period == 0) { // Non-repeating, remove
queue.removeMin();
task.state = TimerTask.EXECUTED;
} else { // Repeating task, reschedule
queue.rescheduleMin(
task.period<0 ? currentTime - task.period
: executionTime + task.period);
}
}
}
if (!taskFired) // Task hasn't yet fired; wait
queue.wait(executionTime - currentTime);
}
if (taskFired) // Task fired; run it, holding no locks
task.run();
} catch(InterruptedException e) {
}
}
}
}我們看到，執行了mainLoop(),裡面是while (true)方法無限循環，取得程式中任務物件中的時間和當前時間比對，相同就執行，但是一旦報錯，就會進入finally中清除掉所有任務資訊。
這時候我們已經找到了答案，timer是在被實例化後，啟動一個線程，不間斷的循環匹配，來執行任務，他是單線程的，一旦報錯，線程就終止了，所以不會執行後續的任務，而ScheduledThreadPoolExecutor是多執行緒執行的，就算其中有一個任務報錯了，並不影響其他執行緒的執行。
2 使用ScheduledThreadPoolExecutor
#從上面看，使用ScheduledThreadPoolExecutor還是比較簡單的，但是我們要實現的更優雅一些，所以選擇TaskScheduler來實作
@Component
public class CronTaskRegistrar implements DisposableBean {
private final Map<Runnable, ScheduledTask> scheduledTasks = new ConcurrentHashMap<>(16);
@Autowired
private TaskScheduler taskScheduler;
public TaskScheduler getScheduler() {
return this.taskScheduler;
}
public void addCronTask(Runnable task, String cronExpression) {
addCronTask(new CronTask(task, cronExpression));
}
private void addCronTask(CronTask cronTask) {
if (cronTask != null) {
Runnable task = cronTask.getRunnable();
if (this.scheduledTasks.containsKey(task)) {
removeCronTask(task);
}
this.scheduledTasks.put(task, scheduleCronTask(cronTask));
}
}
public void removeCronTask(Runnable task) {
Set<Runnable> runnables = this.scheduledTasks.keySet();
Iterator it1 = runnables.iterator();
while (it1.hasNext()) {
SchedulingRunnable schedulingRunnable = (SchedulingRunnable) it1.next();
Long taskId = schedulingRunnable.getTaskId();
SchedulingRunnable cancelRunnable = (SchedulingRunnable) task;
if (taskId.equals(cancelRunnable.getTaskId())) {
ScheduledTask scheduledTask = this.scheduledTasks.remove(schedulingRunnable);
if (scheduledTask != null){
scheduledTask.cancel();
}
}
}
}
public ScheduledTask scheduleCronTask(CronTask cronTask) {
ScheduledTask scheduledTask = new ScheduledTask();
scheduledTask.future = this.taskScheduler.schedule(cronTask.getRunnable(), cronTask.getTrigger());
return scheduledTask;
}
@Override
public void destroy() throws Exception {
for (ScheduledTask task : this.scheduledTasks.values()) {
task.cancel();
}
this.scheduledTasks.clear();
}
}TaskScheduler是本次功能實作的核心類，但是他是一個介面
public interface TaskScheduler {
/**
* Schedule the given {@link Runnable}, invoking it whenever the trigger
* indicates a next execution time.
* <p>Execution will end once the scheduler shuts down or the returned
* {@link ScheduledFuture} gets cancelled.
* @param task the Runnable to execute whenever the trigger fires
* @param trigger an implementation of the {@link Trigger} interface,
* e.g. a {@link org.springframework.scheduling.support.CronTrigger} object
* wrapping a cron expression
* @return a {@link ScheduledFuture} representing pending completion of the task,
* or {@code null} if the given Trigger object never fires (i.e. returns
* {@code null} from {@link Trigger#nextExecutionTime})
* @throws org.springframework.core.task.TaskRejectedException if the given task was not accepted
* for internal reasons (e.g. a pool overload handling policy or a pool shutdown in progress)
* @see org.springframework.scheduling.support.CronTrigger
*/
@Nullable
ScheduledFuture<?> schedule(Runnable task, Trigger trigger);前面的程式碼可以看到，我們在類別中註入了這個類，但是他是接口，我們怎麼知道是那個實現類呢，以往出現這種情況要在類上面加@Primany或者@Quality來執行實現的類，但是我們看到我的注入上並沒有標記，因為是透過另一種方式實現的
@Configuration
public class SchedulingConfig {
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler();
// 定时任务执行线程池核心线程数
taskScheduler.setPoolSize(4);
taskScheduler.setRemoveOnCancelPolicy(true);
taskScheduler.setThreadNamePrefix("TaskSchedulerThreadPool-");
return taskScheduler;
}
}在spring初始化時就註冊了Bean TaskScheduler，而我們可以看到他的實作是ThreadPoolTaskScheduler，在網路上的資料中有人說ThreadPoolTaskScheduler是TaskScheduler的預設實作類，其實不是，還是需要我們去指定，而這種方式，當我們想要替換實作時，只需要修改配置類別就行了，很靈活。
而為什麼說他是更優雅的實現方式呢，因為他的核心也是透過ScheduledThreadPoolExecutor來實現的
public ScheduledExecutorService getScheduledExecutor() throws IllegalStateException {
Assert.state(this.scheduledExecutor != null, "ThreadPoolTaskScheduler not initialized");
return this.scheduledExecutor;
}三多節點任務執行問題
這次的在實作過程中，我並沒有選擇xxl-job來實現，而是採用了TaskScheduler來實現，這也產生了一個問題，xxl-job是分散式的程式調度系統，當想要執行定時任務的應用使用xxl-job時，無論應用程式中部署多少個節點，xxl-job只會選擇其中一個節點作為定時任務執行的節點，從而不會產生定時任務在不同節點上同時執行，導致重複執行問題，而使用TaskScheduler來實現，就要考慮多節點重複執行問題。當然既然有問題，就有解決方案
· 方案一將定時任務功能拆出來單獨部署，且只部署一個節點· 方案二使用redis setNx的形式，保證同一時間只有一個任務在執行
我選擇的是方案二來執行，當然還有一些方式也能保證不重複執行，這裡就不多說了，一下是我的實作
public void executeTask(Long taskId) {
if (!redisService.setIfAbsent(String.valueOf(taskId),"1",2L, TimeUnit.SECONDS)) {
log.info("已有执行中定时发送短信任务，本次不执行！");
return;
}以上是SpringBoot定時任務功能怎麼實現的詳細內容。更多資訊請關注PHP中文網其他相關文章！
陳述：
本文轉載於：yisu.com。如有侵權，請聯絡admin@php.cn刪除