package com.trafficaudit.reportexport.service;
|
|
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
|
import com.trafficaudit.dataimport.entity.ImportBatch;
|
import com.trafficaudit.dataimport.mapper.ImportBatchMapper;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.PreDestroy;
|
import javax.annotation.Resource;
|
import java.util.Date;
|
import java.util.Iterator;
|
import java.util.LinkedHashMap;
|
import java.util.List;
|
import java.util.Map;
|
import java.util.UUID;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ExecutorService;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.TimeUnit;
|
|
/**
|
* 报表导出异步任务：把耗时十几秒的《生成_道路运输量汇总表》生成挪到后台线程，
|
* 前端只拿「任务号 + 进度（步骤/百分比）」，完成后单独下载文件。
|
*
|
* 目的（2026-09-20 用户上报「点下载要等一会，数据多了会不会卡死」）：
|
* 1) 点击后立刻有反馈，不再长时间无响应；
|
* 2) 生成过程不占用 HTTP 请求线程，避免多人同时点把容器拖死（工作线程固定 1 个，多余请求显示「排队中」）；
|
* 3) 同一报表期 10 分钟内的结果可复用，避免每次重算整本约 4 万格公式。
|
*
|
* 缓存正确性守卫：只有当「距上次生成不超过 TTL」且「期间没有任何新的数据导入（import_batch 最大 id 未变）」
|
* 时才复用；导入即失效，避免把过期报表给用户。前端仍提供「重新生成」强制刷新。
|
*/
|
@Slf4j
|
@Service
|
public class ReportExportTaskService {
|
|
/** 汇总工作簿导出任务类型 */
|
public static final String TYPE_SUMMARY_WORKBOOK = "summaryWorkbook";
|
|
/** 生成结果保留时长：期间可反复下载，过后自动清理 */
|
private static final long RESULT_TTL_MS = 30 * 60 * 1000L;
|
/** 同一报表期结果复用时长（且期间无新导入时才复用） */
|
private static final long CACHE_TTL_MS = 10 * 60 * 1000L;
|
|
@Resource
|
private ReportExportService reportService;
|
@Resource
|
private ImportBatchMapper importBatchMapper;
|
|
private final Map<String, Task> tasks = new ConcurrentHashMap<>();
|
/** 同报表期最近一次成功结果：key -> task */
|
private final Map<String, Task> cache = new ConcurrentHashMap<>();
|
private final ExecutorService pool = Executors.newFixedThreadPool(1, r -> {
|
Thread t = new Thread(r, "report-export-task");
|
t.setDaemon(true);
|
return t;
|
});
|
|
/** 一次导出任务（内存态，进程重启即失效） */
|
public static class Task {
|
private final String id = UUID.randomUUID().toString().replace("-", "");
|
private final String type;
|
private final String period;
|
private final String mode;
|
private final Integer toMonth;
|
private final long createdAt = System.currentTimeMillis();
|
/** 生成时的数据版本（import_batch 最大 id），用于判断缓存是否过期 */
|
private final long importVersion;
|
private volatile String status = "PENDING"; // PENDING/RUNNING/DONE/FAILED
|
private volatile int percent = 0;
|
private volatile String step = "排队中";
|
private volatile String error;
|
private volatile byte[] data;
|
private volatile String fileName;
|
private volatile boolean fromCache;
|
private volatile long finishedAt;
|
|
Task(String type, String period, String mode, Integer toMonth, long importVersion) {
|
this.type = type;
|
this.period = period;
|
this.mode = mode;
|
this.toMonth = toMonth;
|
this.importVersion = importVersion;
|
}
|
|
void progress(int pct, String stepText) {
|
if (pct > this.percent) this.percent = Math.min(pct, 99);
|
if (stepText != null) this.step = stepText;
|
}
|
|
public String getId() { return id; }
|
|
public String getStatus() { return status; }
|
|
public boolean isDone() { return "DONE".equals(status) && data != null; }
|
|
public boolean isFromCache() { return fromCache; }
|
|
public byte[] getData() { return data; }
|
|
public String getFileName() { return fileName; }
|
|
public String getError() { return error; }
|
|
public Map<String, Object> toView() {
|
Map<String, Object> m = new LinkedHashMap<>();
|
m.put("taskId", id);
|
m.put("type", type);
|
m.put("period", period);
|
m.put("mode", mode);
|
m.put("toMonth", toMonth);
|
m.put("status", status);
|
m.put("percent", "DONE".equals(status) ? 100 : percent);
|
m.put("step", step);
|
m.put("fromCache", fromCache);
|
m.put("fileName", fileName);
|
m.put("error", error);
|
long end = finishedAt > 0 ? finishedAt : System.currentTimeMillis();
|
m.put("elapsedMs", end - createdAt);
|
m.put("finishedAt", finishedAt > 0 ? new Date(finishedAt) : null);
|
return m;
|
}
|
}
|
|
/**
|
* 提交导出任务。forceRefresh=true 时忽略缓存重新生成；
|
* 否则在「同一报表期、10 分钟内、期间无新导入」的前提下直接复用上次结果。
|
* 同一报表期已有任务在跑时直接返回该任务，避免重复排队。
|
*/
|
public Task submit(String type, String period, String mode, Integer toMonth, boolean forceRefresh) {
|
cleanupExpired();
|
String key = cacheKey(type, period, mode, toMonth);
|
if (!forceRefresh) {
|
Task hit = cache.get(key);
|
if (cacheUsable(hit)) {
|
hit.fromCache = true;
|
log.info("报表导出任务命中缓存：{} {}（生成于 {}）", type, period, new Date(hit.finishedAt));
|
return hit;
|
}
|
}
|
Task running = findRunning(key);
|
if (running != null) {
|
log.info("报表导出任务已在执行，复用进行中的任务：{} {}（{}）", type, period, running.id);
|
return running;
|
}
|
Task task = new Task(type, period, mode, toMonth, importVersion());
|
tasks.put(task.id, task);
|
pool.submit(() -> run(task));
|
return task;
|
}
|
|
public Task get(String taskId) {
|
if (taskId == null) return null;
|
cleanupExpired();
|
return tasks.get(taskId);
|
}
|
|
private void run(Task t) {
|
t.status = "RUNNING";
|
t.step = "开始生成";
|
long start = System.currentTimeMillis();
|
try {
|
byte[] data = reportService.exportSummaryWorkbook(t.period, t.mode, t.toMonth, t::progress);
|
t.data = data;
|
t.fileName = "生成_道路运输量汇总表_" + t.period + ".xlsx";
|
t.step = "完成";
|
t.status = "DONE";
|
t.finishedAt = System.currentTimeMillis();
|
cache.put(cacheKey(t.type, t.period, t.mode, t.toMonth), t);
|
log.info("报表导出任务完成：{} {} 用时 {} ms，{} 字节", t.type, t.period, t.finishedAt - start, data.length);
|
} catch (Throwable e) {
|
t.status = "FAILED";
|
t.error = e.getMessage() == null ? e.toString() : e.getMessage();
|
t.step = "生成失败";
|
t.finishedAt = System.currentTimeMillis();
|
log.error("报表导出任务失败：{} {}（用时 {} ms）", t.type, t.period, t.finishedAt - start, e);
|
}
|
}
|
|
private Task findRunning(String key) {
|
for (Task t : tasks.values()) {
|
if (!key.equals(cacheKey(t.type, t.period, t.mode, t.toMonth))) continue;
|
if ("PENDING".equals(t.status) || "RUNNING".equals(t.status)) return t;
|
}
|
return null;
|
}
|
|
private boolean cacheUsable(Task t) {
|
if (t == null || !"DONE".equals(t.status) || t.data == null) return false;
|
if (System.currentTimeMillis() - t.finishedAt > CACHE_TTL_MS) return false;
|
long now = importVersion();
|
if (now < 0 || t.importVersion < 0) return false; // 取不到数据版本时不冒风险复用
|
return now == t.importVersion;
|
}
|
|
/** 数据版本＝import_batch 最大 id：任何一次导入都会让它变化，取不到返回 -1（此时禁用缓存） */
|
private long importVersion() {
|
try {
|
List<Object> v = importBatchMapper.selectObjs(new QueryWrapper<ImportBatch>().select("IFNULL(MAX(id),0)"));
|
if (v == null || v.isEmpty() || v.get(0) == null) return 0L;
|
return ((Number) v.get(0)).longValue();
|
} catch (Exception e) {
|
log.warn("读取导入数据版本失败，本次禁用导出缓存：{}", e.getMessage());
|
return -1L;
|
}
|
}
|
|
private String cacheKey(String type, String period, String mode, Integer toMonth) {
|
return type + "|" + period + "|" + mode + "|" + (toMonth == null ? "" : toMonth);
|
}
|
|
private void cleanupExpired() {
|
long now = System.currentTimeMillis();
|
for (Iterator<Map.Entry<String, Task>> it = tasks.entrySet().iterator(); it.hasNext(); ) {
|
Task t = it.next().getValue();
|
if (t.finishedAt > 0 && now - t.finishedAt > RESULT_TTL_MS) it.remove();
|
}
|
for (Iterator<Map.Entry<String, Task>> it = cache.entrySet().iterator(); it.hasNext(); ) {
|
Task t = it.next().getValue();
|
if (t.finishedAt <= 0 || now - t.finishedAt > CACHE_TTL_MS) it.remove();
|
}
|
}
|
|
@PreDestroy
|
public void shutdown() {
|
pool.shutdownNow();
|
try {
|
pool.awaitTermination(3, TimeUnit.SECONDS);
|
} catch (InterruptedException e) {
|
Thread.currentThread().interrupt();
|
}
|
}
|
}
|
