1. Summary
Speaking of timing tasks, no one knows how to pattern it. I saw Java a few days agoTimerFor timed (delay) tasks, follow the frameworks of Timer - ScheduledThreadPoolExecutor - RocketMQ - Netty, I have read some core logical implementations of timed tasks. Now I will write a series of articles to record the implementation process of these source codes. If there are any errors, please point them out.
Timing tasks are more or less exposed to our daily projects, and are very useful in many scenarios, such as:
- Process data in batches, generate reports, etc. every day
- Meeting reminders, email sending, etc.
- Order payment timeout refund, cleaning of temporary files, log files...
Before we get to the point, we will implement the simplest timer/delay ourselves.
2. Pay attention to the problem
To implement a timed task, we must first consider the following issues
- How to sort tasks by execution time (what kind of collection)
- How to execute the scheduled task after it expires
- How to implement task execution multiple times
With these three questions in mind, let’s consider the implementation of timer
First of all, what kind of sets can meet our requirements? It needs to be automatically sorted and adjusted in a fixed order. Just use the JDK PriorityQueue priority queue. You only need to set the sorting according to the execution time of the task. When we add the task to the queue, the queue will be automatically sorted according to the execution time of the task. We don’t need to consider this part of the code.
The second problem is that we can start an additional task processing thread to specifically loop the queue to process the expired tasks. If the task does not expire, it will block and wait, and if it expires, it will wake up the execution of the task.
The third question is, when adding a task, you can set a Count to indicate the number of times the task needs to be executed. When adding a task, you can loop through the addition
3. Implementation
3.1 Task definition
First, define a task bean, which includes the execution time of the task and the executed tasks:
class Task {
// Execution time private long executeTime;
// Task Runnable runnable;
public Task(long executeTime, Runnable runnable) {
= executeTime;
= runnable;
}
public long getExecuteTime() {
return executeTime;
}
public void setExecuteTime(long executeTime) {
= executeTime;
}
public Runnable getRunnable() {
return runnable;
}
public void setRunnable(Runnable runnable) {
= runnable;
}
}
3.2 Methods to add tasks
public boolean add(long addTime, Runnable runnable, int count) {
long now = ();
synchronized (queue) {
for (int i = 0; i < count; i++) {
// Calculate the execution time of the task long newTime = now + addTime * (i + 1);
// If it is less than now, it means it is overflowing if (newTime < now) {
throw new RuntimeException(().getName() + ": overflow long limit");
}
Task min = getMin();
if (min == null || newTime < ) {
// Join the task queue (new Task(newTime, runnable));
// Wake up the task thread ();
continue;
}
// Join the task queue (new Task(newTime, runnable));
}
}
return true;
}First, calculate the execution time of the task and determine whether it will overflow. If it overflows, an exception will be thrown.
Then get the team's first task. If the execution time of the added new task is more blocky than the team's first task, call () to wake up the thread to execute the task.
Add tasks to queue
3.3 Task thread
Thread thread = new Thread(() -> {
(().getName() + ": Start the execute-job-thread thread");
while (true) {
try {
synchronized (queue) {
if (()) {
();
}
long now = ();
while (!() && now >= ().getExecuteTime()) {
Task task = ();
// Thread pool executes tasks ();
}
if (()) {
();
} else {
long div = ().executeTime - now;
(div);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}, "execute-job-thread");The thread keeps traversing the queue in a dead loop to see if there is any task value
- If there is no task, block and wait
- After being awakened, there is a task. At this time, you can get the current time and get all expired tasks to execute in turn.
- After execution, we will determine how long it will be blocked based on whether there are still tasks in the queue.
3.4 Overall code
The PQueueService code is as follows:
public class PQueueService {
private static final int MAX_COUNT = Integer.MAX_VALUE;
private final PriorityQueue<Task> queue = new PriorityQueue<>((o1, o2) -> {
return (, );
});
static final ThreadPoolExecutor executor =
new ThreadPoolExecutor(16, 32, 1, , new LinkedBlockingQueue<>(16));
Thread thread = new Thread(() -> {
(().getName() + ": Start the execute-job-thread thread");
while (true) {
try {
synchronized (queue) {
if (()) {
();
}
long now = ();
while (!() && now >= ().getExecuteTime()) {
Task task = ();
// Thread pool executes tasks ();
}
if (()) {
();
} else {
long div = ().executeTime - now;
(div);
}
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}, "execute-job-thread");
public void init() {
();
}
public Task getMin() {
if (()) {
return null;
}
synchronized (queue) {
return ();
}
}
public Task pollMin() {
if (()) {
return null;
}
synchronized (queue) {
return ();
}
}
public boolean add(long addTime, Runnable runnable, int count) {
long now = ();
synchronized (queue) {
for (int i = 0; i < count; i++) {
long newTime = now + addTime * (i + 1);
if (newTime < now) {
throw new RuntimeException(().getName() + ": overflow long limit");
}
Task min = getMin();
if (min == null || newTime < ) {
// Join the task queue (new Task(newTime, runnable));
// Wake up the task thread ();
continue;
}
// Join the task queue (new Task(newTime, runnable));
}
}
return true;
}
class Task {
private long executeTime;
Runnable runnable;
public Task(long executeTime, Runnable runnable) {
= executeTime;
= runnable;
}
public long getExecuteTime() {
return executeTime;
}
public void setExecuteTime(long executeTime) {
= executeTime;
}
public Runnable getRunnable() {
return runnable;
}
public void setRunnable(Runnable runnable) {
= runnable;
}
}
}
The Timer class code is as follows:
public class Timer {
public static final Logger logger = ();
PQueueService pQueueService;
public Timer() {
pQueueService = new PQueueService();
();
}
/**
* Perform tasks
*/
public void fixExecuteOne(int seconds, TimeUnit timeUnit, Runnable runnable, int count) {
if (() != ()) {
throw new RuntimeException("unknown unit");
}
long time = seconds * 1000L;
(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. Add a task that will be executed after 10s (10, , () -> {
("10sTask ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
// 2. Add a task that will be executed after 3s (3, , () -> {
("3sTask ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
// 3. Add a task that will be executed after 20s (20, , () -> {
("20sTask ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
// 4. Add a task that will be executed after 1s (1, , () -> {
("1sTask1 ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
// 5. Add a task that will be executed after 1s (1, , () -> {
("1sTask2 ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
// 6. Add a task that will be executed after 1s (1, , () -> {
("1sTask3 ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 2);
try {
(30000);
// 7. Add a task that will be executed after 1s (1, , () -> {
("1sTask3 ===> " + ().getName() + ": " + "Execute!!!, Current time:" + (new Date()));
}, 1);
(30000);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
}
Execution results:
execute-job-thread: Start the execute-job-thread thread
1s Task 2 ===> pool-1-thread-2: Execute!!!, Current time: 2024-11-23 21:17:21
1s task 1 ===> pool-1-thread-1: Execute!!!, Current time: 2024-11-23 21:17:21
1s Task 3 ===> pool-1-thread-3: Execution!!!, Current time: 2024-11-23 21:17:21
1s task 1 ===> pool-1-thread-4: Execute!!!, Current time: 2024-11-23 21:17:22
1s Task 2 ===> pool-1-thread-5: Execute!!!, Current time: 2024-11-23 21:17:22
1s Task 3 ===> pool-1-thread-6: Execution!!!, Current time: 2024-11-23 21:17:22
3s task ===> pool-1-thread-7: Execute!!!, Current time: 2024-11-23 21:17:23
3s task ===> pool-1-thread-8: Execution!!!, Current time: 2024-11-23 21:17:26
10s task ===> pool-1-thread-9: Execution!!!, Current time: 2024-11-23 21:17:30
10s task ===> pool-1-thread-10: Execute!!!, Current time: 2024-11-23 21:17:40
20s task ===> pool-1-thread-11: Execute!!!, Current time: 2024-11-23 21:17:40
1s task added after 30s ===> pool-1-thread-12: Execute!!!, Current time: 2024-11-23 21:17:51
20s task ===> pool-1-thread-13: Execution!!!, Current time: 2024-11-23 21:18:00
4. There are problems
A simple timer is implemented above, but this timer still has the following problems:
- The function is relatively simple, no other remove or other operations
- After the task thread receives the task, it will be thrown into another thread pool for execution. If there are many tasks in the thread pool, it will block and lead to inaccurate execution time.
- The queue is locked in the code, but a task can be added to multiple queues, and there is no concurrency processing here
- …
In short, this is a simple Timer Demo, which mainly understands the execution process of timers/delays, but in fact, the implementation process of timers/delays in JDK is very different. In the next article, we will take a look at the usage and source code of Timer.
This is the end of this article about implementing a simple timer using Java. For more related Java timer content, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!