stu/asyncTool
1
0
Fork 0
mirror of https://gitee.com/jd-platform-opensource/asyncTool.git synced 2026-03-22 04:27:15 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

超时时间轮---完成

Browse Source
This commit is contained in:
shenkaiwen5
2021-11-16 17:45:35 +08:00
parent 47f707f287
commit d41c09383b
2 changed files with 8 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
src/main/java/com/jd/platform/async/executor/wheel/TimeWheel.java
View File

@@ -85,7 +85,6 @@ public class TimeWheel {
*/
public boolean addTask(TimerTask timerTask, long currentTime) {
long expiration = timerTask.getDelayMsAndCur();
currentTime = currentTime - (currentTime % tickMs);
//过期任务直接执行
if (expiration < currentTime + tickMs) {
@@ -97,14 +96,14 @@ public class TimeWheel {
//System.out.println("tickMs:" + tickMs + "------index:" + index + "------expiration:" + expiration);
TimerTaskList timerTaskList = timerTaskLists[index];
timerTaskList.addTask(timerTask);
if (timerTaskList.setExpiration(virtualId * tickMs)) {
//加到delayList对应元素 【20槽, 时间轮层数】,这里添加前一定要初始化到对应层数
while (delayList.get(index).size() < overflowIndex + 1) {
delayList.get(index).add(new TimerTaskList());
}
//这里每次清空整个槽位的List，并且flush List的元素链表，所以没问题
delayList.get(index).set(overflowIndex, timerTaskList);
//加到delayList对应元素 【20槽, 时间轮层数】,这里添加前一定要初始化到对应层数
while (delayList.get(index).size() < overflowIndex + 1) {
delayList.get(index).add(new TimerTaskList());
}
//这里每次清空整个槽位的List，并且flush List的元素链表，所以没问题
delayList.get(index).set(overflowIndex, timerTaskList);
} else {
//放到上一层的时间轮
TimeWheel timeWheel = getOverflowWheel();

38
src/main/java/com/jd/platform/async/executor/wheel/TimerTaskList.java
View File

@@ -1,19 +1,11 @@
package com.jd.platform.async.executor.wheel;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
/**
* 时间槽
*/
public class TimerTaskList implements Delayed {
/**
* 过期时间
*/
private AtomicLong expiration = new AtomicLong(-1L);
public class TimerTaskList {
/**
* 根节点
@@ -25,20 +17,6 @@ public class TimerTaskList implements Delayed {
root.next = root;
}
/**
* 设置过期时间
*/
public boolean setExpiration(long expire) {
return expiration.getAndSet(expire) != expire;
}
/**
* 获取过期时间
*/
public long getExpiration() {
return expiration.get();
}
/**
* 新增任务
*/
@@ -80,19 +58,5 @@ public class TimerTaskList implements Delayed {
flush.accept(timerTask);
timerTask = root.next;
}
expiration.set(-1L);
}
@Override
public long getDelay(TimeUnit unit) {
return Math.max(0, unit.convert(expiration.get() - System.currentTimeMillis(), TimeUnit.MILLISECONDS));
}
@Override
public int compareTo(Delayed o) {
if (o instanceof TimerTaskList) {
return Long.compare(expiration.get(), ((TimerTaskList) o).expiration.get());
}
return 0;
}
}