package com.zy.core.task;
|
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Component;
|
|
import javax.annotation.PreDestroy;
|
import java.util.Map;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.LinkedBlockingQueue;
|
import java.util.concurrent.ThreadFactory;
|
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.atomic.AtomicBoolean;
|
import java.util.concurrent.atomic.AtomicInteger;
|
|
/**
|
* 主流程异步任务调度器。
|
* 按执行通道串行执行任务，并对同名任务提供最小提交间隔和不重入保护。
|
*/
|
@Slf4j
|
@Component
|
public class MainProcessAsyncTaskScheduler {
|
|
private static final long DEFAULT_SHUTDOWN_TIMEOUT_SECONDS = 5L;
|
private static final long IDLE_LANE_TTL_MS = 60_000L;
|
private static final long STALE_TASK_GUARD_TTL_MS = 60_000L;
|
|
private final Map<String, SerialTaskLane> laneMap = new ConcurrentHashMap<>();
|
private final Map<String, TaskGuard> taskGuardMap = new ConcurrentHashMap<>();
|
|
public boolean submit(String laneName,
|
String taskName,
|
long minIntervalMs,
|
long slowLogThresholdMs,
|
Runnable task) {
|
if (isBlank(laneName) || isBlank(taskName) || task == null) {
|
return false;
|
}
|
|
long now = System.currentTimeMillis();
|
String taskKey = buildTaskKey(laneName, taskName);
|
TaskGuard taskGuard = taskGuardMap.computeIfAbsent(taskKey, key -> new TaskGuard());
|
synchronized (taskGuard) {
|
if (taskGuard.lastSubmitTimeMs > 0L && now - taskGuard.lastSubmitTimeMs < minIntervalMs) {
|
return false;
|
}
|
if (!taskGuard.running.compareAndSet(false, true)) {
|
return false;
|
}
|
taskGuard.lastSubmitTimeMs = now;
|
taskGuard.lastQueueEnterTimeMs = now;
|
}
|
|
try {
|
cleanupStaleState(now);
|
SerialTaskLane lane = ensureLane(laneName);
|
lane.touch(now);
|
lane.executorService.execute(() -> executeTask(laneName, taskName, slowLogThresholdMs, taskGuard, task));
|
return true;
|
} catch (Exception e) {
|
taskGuard.running.set(false);
|
log.error("MainProcess async task submit error, lane={}, task={}", laneName, taskName, e);
|
return false;
|
}
|
}
|
|
private void executeTask(String laneName,
|
String taskName,
|
long slowLogThresholdMs,
|
TaskGuard taskGuard,
|
Runnable task) {
|
long startMs = System.currentTimeMillis();
|
long queueWaitMs = taskGuard.lastQueueEnterTimeMs > 0L ? startMs - taskGuard.lastQueueEnterTimeMs : 0L;
|
try {
|
task.run();
|
} catch (Exception e) {
|
log.error("MainProcess async task execute error, lane={}, task={}", laneName, taskName, e);
|
} finally {
|
long costMs = System.currentTimeMillis() - startMs;
|
if (slowLogThresholdMs > 0L && costMs > slowLogThresholdMs) {
|
log.warn("MainProcess async task executed slowly, lane={}, task={}, cost={}ms, queueWaitMs={}ms", laneName, taskName, costMs, queueWaitMs);
|
}
|
taskGuard.running.set(false);
|
cleanupTaskGuardIfIdle(buildTaskKey(laneName, taskName), taskGuard, System.currentTimeMillis());
|
}
|
}
|
|
private SerialTaskLane ensureLane(String laneName) {
|
SerialTaskLane lane = laneMap.get(laneName);
|
if (lane != null && !lane.executorService.isShutdown()) {
|
return lane;
|
}
|
synchronized (laneMap) {
|
lane = laneMap.get(laneName);
|
if (lane == null || lane.executorService.isShutdown()) {
|
lane = new SerialTaskLane(createLaneExecutor(laneName));
|
laneMap.put(laneName, lane);
|
}
|
return lane;
|
}
|
}
|
|
private ThreadPoolExecutor createLaneExecutor(String laneName) {
|
ThreadPoolExecutor executor = new ThreadPoolExecutor(
|
1,
|
1,
|
IDLE_LANE_TTL_MS,
|
TimeUnit.MILLISECONDS,
|
new LinkedBlockingQueue<>(),
|
new NamedThreadFactory("main-process-" + laneName + "-")
|
);
|
executor.allowCoreThreadTimeOut(true);
|
return executor;
|
}
|
|
private void cleanupStaleState(long now) {
|
cleanupStaleTaskGuards(now);
|
cleanupIdleLanes(now);
|
}
|
|
private void cleanupStaleTaskGuards(long now) {
|
for (Map.Entry<String, TaskGuard> entry : taskGuardMap.entrySet()) {
|
cleanupTaskGuardIfIdle(entry.getKey(), entry.getValue(), now);
|
}
|
}
|
|
private void cleanupTaskGuardIfIdle(String taskKey, TaskGuard taskGuard, long now) {
|
if (taskGuard == null || taskGuard.running.get()) {
|
return;
|
}
|
long idleMs = now - taskGuard.lastSubmitTimeMs;
|
if (taskGuard.lastSubmitTimeMs <= 0L || idleMs < STALE_TASK_GUARD_TTL_MS) {
|
return;
|
}
|
taskGuardMap.remove(taskKey, taskGuard);
|
}
|
|
private void cleanupIdleLanes(long now) {
|
synchronized (laneMap) {
|
for (Map.Entry<String, SerialTaskLane> entry : laneMap.entrySet()) {
|
SerialTaskLane lane = entry.getValue();
|
if (lane == null || !lane.isIdle(now)) {
|
continue;
|
}
|
if (laneMap.remove(entry.getKey(), lane)) {
|
lane.executorService.shutdown();
|
}
|
}
|
}
|
}
|
|
@PreDestroy
|
public void shutdown() {
|
for (SerialTaskLane lane : laneMap.values()) {
|
if (lane == null || lane.executorService == null) {
|
continue;
|
}
|
lane.executorService.shutdownNow();
|
}
|
for (Map.Entry<String, SerialTaskLane> entry : laneMap.entrySet()) {
|
SerialTaskLane lane = entry.getValue();
|
if (lane == null || lane.executorService == null) {
|
continue;
|
}
|
try {
|
if (!lane.executorService.awaitTermination(DEFAULT_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
|
log.warn("MainProcess async task lane did not terminate in time, lane={}", entry.getKey());
|
}
|
} catch (InterruptedException e) {
|
Thread.currentThread().interrupt();
|
return;
|
}
|
}
|
}
|
|
private String buildTaskKey(String laneName, String taskName) {
|
return laneName + "::" + taskName;
|
}
|
|
private boolean isBlank(String value) {
|
return value == null || value.trim().isEmpty();
|
}
|
|
private static class SerialTaskLane {
|
|
private final ThreadPoolExecutor executorService;
|
private volatile long lastTouchTimeMs = System.currentTimeMillis();
|
|
private SerialTaskLane(ThreadPoolExecutor executorService) {
|
this.executorService = executorService;
|
}
|
|
private void touch(long now) {
|
this.lastTouchTimeMs = now;
|
}
|
|
private boolean isIdle(long now) {
|
if (executorService.isShutdown()) {
|
return true;
|
}
|
if (executorService.getActiveCount() > 0) {
|
return false;
|
}
|
if (!executorService.getQueue().isEmpty()) {
|
return false;
|
}
|
return now - lastTouchTimeMs >= IDLE_LANE_TTL_MS;
|
}
|
}
|
|
private static class TaskGuard {
|
|
private final AtomicBoolean running = new AtomicBoolean(false);
|
private volatile long lastSubmitTimeMs = 0L;
|
private volatile long lastQueueEnterTimeMs = 0L;
|
}
|
|
private static class NamedThreadFactory implements ThreadFactory {
|
|
private final String prefix;
|
private final AtomicInteger index = new AtomicInteger(1);
|
|
private NamedThreadFactory(String prefix) {
|
this.prefix = prefix;
|
}
|
|
@Override
|
public Thread newThread(Runnable runnable) {
|
Thread thread = new Thread(runnable, prefix + index.getAndIncrement());
|
thread.setDaemon(true);
|
return thread;
|
}
|
}
|
}
|
