|
@@ -14,9 +14,11 @@ import org.springframework.context.event.ContextRefreshedEvent;
|
|
|
import org.springframework.context.event.EventListener;
|
|
import org.springframework.context.event.EventListener;
|
|
|
import org.springframework.stereotype.Service;
|
|
import org.springframework.stereotype.Service;
|
|
|
|
|
|
|
|
|
|
+import java.util.Comparator;
|
|
|
import java.util.concurrent.DelayQueue;
|
|
import java.util.concurrent.DelayQueue;
|
|
|
import java.util.concurrent.ExecutorService;
|
|
import java.util.concurrent.ExecutorService;
|
|
|
import java.util.concurrent.Executors;
|
|
import java.util.concurrent.Executors;
|
|
|
|
|
+import java.util.stream.Stream;
|
|
|
|
|
|
|
|
/**
|
|
/**
|
|
|
* @author skyline
|
|
* @author skyline
|
|
@@ -78,8 +80,10 @@ public class ActivityDelayJob implements DelayService<DelayActivity> {
|
|
|
.setType(DelayActivity.TYPE_结束))
|
|
.setType(DelayActivity.TYPE_结束))
|
|
|
.toList();
|
|
.toList();
|
|
|
|
|
|
|
|
- delayStartActivityList.stream().map(delay -> new DelayedItem<>(delay, delay.getExecuteTime())).toList().forEach(this::addToDelayQueue);
|
|
|
|
|
- delayEndActivityList.stream().map(delay -> new DelayedItem<>(delay, delay.getExecuteTime())).toList().forEach(this::addToDelayQueue);
|
|
|
|
|
|
|
+ // 启动队列和结束队列合并之后按照时间排序加入队列
|
|
|
|
|
+ Stream.concat(delayStartActivityList.stream(), delayEndActivityList.stream())
|
|
|
|
|
+ .sorted(Comparator.comparing(DelayActivity::getExecuteTime))
|
|
|
|
|
+ .map(delay -> new DelayedItem<>(delay, delay.getExecuteTime())).toList().forEach(this::addToDelayQueue);
|
|
|
|
|
|
|
|
// 开启线程处理队列消息
|
|
// 开启线程处理队列消息
|
|
|
executor.execute(() -> {
|
|
executor.execute(() -> {
|
|
@@ -152,13 +156,13 @@ public class ActivityDelayJob implements DelayService<DelayActivity> {
|
|
|
|
|
|
|
|
@Override
|
|
@Override
|
|
|
public boolean addToDelayQueue(DelayedItem<DelayActivity> delayedItem) {
|
|
public boolean addToDelayQueue(DelayedItem<DelayActivity> delayedItem) {
|
|
|
- return DELAY_QUEUE.add(delayedItem);
|
|
|
|
|
|
|
+ return DELAY_QUEUE.offer(delayedItem);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
@Override
|
|
@Override
|
|
|
public boolean addToDelayQueue(DelayActivity activity) {
|
|
public boolean addToDelayQueue(DelayActivity activity) {
|
|
|
- DelayedItem<DelayActivity> orderDelayed = new DelayedItem<>(activity, activity.getExecuteTime());
|
|
|
|
|
- return DELAY_QUEUE.add(orderDelayed);
|
|
|
|
|
|
|
+ DelayedItem<DelayActivity> activityDelayed = new DelayedItem<>(activity, activity.getExecuteTime());
|
|
|
|
|
+ return DELAY_QUEUE.offer(activityDelayed);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
|