This commit is contained in:
dzc committed 2026-10-21 19:39:59 +08:00
1 parent c34b249d7f
commit f4c4390352
114 files changed
+8278 -631

No files matched your search

@@ -0,0 +1,95 @@
package com.bsd.pjcloud.common.asynctask.config;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskExecutorService;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
import com.bsd.pjcloud.common.asynctask.service.impl.DefaultAsyncTaskExecutorService;
import com.bsd.pjcloud.common.asynctask.service.impl.DefaultAsyncTaskProgressService;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.context.annotation.Bean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.transaction.support.TransactionTemplate;
/**
* 通用异步任务自动配置。
* <p>业务模块引入本公共模块后,在 Redis 和事务模板均可用时自动装配专用线程池、进度服务与事务执行器。</p>
*/
@AutoConfiguration
@AutoConfigureAfter(name = {
"com.bsd.pjcloud.common.data.cache.RedisTemplateConfiguration",
"org.springframework.boot.autoconfigure.transaction.TransactionAutoConfiguration"
})
@ConditionalOnClass({RedisTemplate.class, TransactionTemplate.class})
@ConditionalOnBean({RedisTemplate.class, TransactionTemplate.class})
public class AsyncTaskAutoConfiguration {
/** 公共异步任务专用工作线程池的 Bean 名称,业务模块可按该名称替换默认线程池。 */
public static final String ASYNC_TASK_WORKER_EXECUTOR_BEAN_NAME = "commonAsyncTaskWorkerExecutor";
/** 默认核心线程数,兼顾常规导入导出并发与业务服务资源占用。 */
private static final int DEFAULT_CORE_POOL_SIZE = 2;
/** 默认最大线程数,限制大批量异步任务同时占用数据库连接和远程调用资源。 */
private static final int DEFAULT_MAX_POOL_SIZE = 8;
/** 默认等待队列容量,超过线程与队列总容量后由执行器拒绝并把任务标记为提交失败。 */
private static final int DEFAULT_QUEUE_CAPACITY = 200;
/** 服务关闭时等待正在执行任务结束的最长秒数,超过后由容器继续关闭流程。 */
private static final int DEFAULT_SHUTDOWN_AWAIT_SECONDS = 30;
/** 公共异步任务线程名称前缀,便于线程转储和日志定位。 */
private static final String DEFAULT_THREAD_NAME_PREFIX = "common-async-task-";
/**
* 装配公共异步任务专用线程池,避免 Spring Boot 默认线程池因其他业务 Executor 存在而不创建。
*
* @return 已完成容量与优雅停机配置的工作线程池
*/
@Bean(name = ASYNC_TASK_WORKER_EXECUTOR_BEAN_NAME)
@ConditionalOnMissingBean(name = ASYNC_TASK_WORKER_EXECUTOR_BEAN_NAME)
public TaskExecutor commonAsyncTaskWorkerExecutor() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(DEFAULT_CORE_POOL_SIZE);
executor.setMaxPoolSize(DEFAULT_MAX_POOL_SIZE);
executor.setQueueCapacity(DEFAULT_QUEUE_CAPACITY);
executor.setThreadNamePrefix(DEFAULT_THREAD_NAME_PREFIX);
executor.setWaitForTasksToCompleteOnShutdown(true);
executor.setAwaitTerminationSeconds(DEFAULT_SHUTDOWN_AWAIT_SECONDS);
return executor;
}
/**
* 装配默认 Redis 进度服务,业务模块可通过声明同类型 Bean 完整替换默认实现。
*
* @param redisTemplate 项目统一配置的 Redis 操作模板
* @return 通用异步任务 Redis 进度服务
*/
@Bean
@ConditionalOnMissingBean
public AsyncTaskProgressService asyncTaskProgressService(RedisTemplate<String, Object> redisTemplate) {
return new DefaultAsyncTaskProgressService(redisTemplate);
}
/**
* 装配默认异步事务执行器,显式使用公共异步任务专用线程池以隔离其他业务线程池。
*
* @param asyncTaskWorkerExecutor 公共异步任务专用工作线程池
* @param asyncTaskProgressService Redis 任务进度服务
* @param transactionTemplate Spring 编程式事务模板
* @return 通用异步事务任务执行器
*/
@Bean
@ConditionalOnMissingBean
public AsyncTaskExecutorService asyncTaskExecutorService(
@Qualifier(ASYNC_TASK_WORKER_EXECUTOR_BEAN_NAME) TaskExecutor asyncTaskWorkerExecutor,
AsyncTaskProgressService asyncTaskProgressService, TransactionTemplate transactionTemplate) {
return new DefaultAsyncTaskExecutorService(asyncTaskWorkerExecutor, asyncTaskProgressService, transactionTemplate);
}
}
@@ -0,0 +1,33 @@
package com.bsd.pjcloud.common.asynctask.config;
import com.bsd.pjcloud.common.asynctask.controller.AsyncTaskProgressController;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
import org.springframework.boot.autoconfigure.AutoConfiguration;
import org.springframework.boot.autoconfigure.AutoConfigureAfter;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnWebApplication;
import org.springframework.context.annotation.Bean;
/**
* 通用异步任务 Servlet Web 接口自动配置。
* <p>仅在 Web 应用已装配 Redis 进度服务时注册统一查询接口,非 Web 任务消费者不会引入无用端点。</p>
*/
@AutoConfiguration
@AutoConfigureAfter(AsyncTaskAutoConfiguration.class)
@ConditionalOnWebApplication(type = ConditionalOnWebApplication.Type.SERVLET)
@ConditionalOnBean(AsyncTaskProgressService.class)
public class AsyncTaskWebAutoConfiguration {
/**
* 装配统一进度查询接口,所有业务模块共享 {@code /async-task/progress} 契约。
*
* @param asyncTaskProgressService Redis 任务进度服务
* @return 通用异步任务进度控制器
*/
@Bean
@ConditionalOnMissingBean
public AsyncTaskProgressController asyncTaskProgressController(AsyncTaskProgressService asyncTaskProgressService) {
return new AsyncTaskProgressController(asyncTaskProgressService);
}
}
@@ -0,0 +1,40 @@
package com.bsd.pjcloud.common.asynctask.constant;
import lombok.Getter;
/**
* 通用异步任务进度状态协议。
* <p>状态编码会写入 Redis 并返回调用端,已有编码不得随意变更。</p>
*/
@Getter
public enum AsyncTaskStatusEnum {
/** 等待后台线程开始执行,Redis 与接口编码为 {@code PENDING}。 */
PENDING("PENDING", "等待执行"),
/** 后台线程正在处理任务,Redis 与接口编码为 {@code RUNNING}。 */
RUNNING("RUNNING", "执行中"),
/** 任务已完整执行成功,Redis 与接口编码为 {@code SUCCESS}。 */
SUCCESS("SUCCESS", "执行成功"),
/** 任务执行异常,Redis 与接口编码为 {@code FAILED}。 */
FAILED("FAILED", "执行失败");
/** Redis 与调用端接口共同使用的固定状态编码。 */
private final String status;
/** 当前状态未提供自定义消息时使用的中文说明。 */
private final String description;
/**
* 创建异步任务状态定义。
*
* @param status Redis 与调用端接口共同使用的固定状态编码
* @param description 当前状态的默认中文说明
*/
AsyncTaskStatusEnum(String status, String description) {
this.status = status;
this.description = description;
}
}
@@ -0,0 +1,50 @@
package com.bsd.pjcloud.common.asynctask.controller;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
import com.bsd.pjcloud.common.core.util.R;
import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag;
import org.springframework.http.HttpHeaders;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
/**
* 通用异步任务进度查询接口。
* <p>任务业务 Key 按登录用户隔离,租户 ID 显式写入服务端任务键,只能查询本人创建且尚未过期的任务。</p>
*/
@RestController
@RequestMapping("/async-task")
@Tag(description = "async-task", name = "通用异步任务进度")
@SecurityRequirement(name = HttpHeaders.AUTHORIZATION)
public class AsyncTaskProgressController {
/** 通用 Redis 任务进度服务。 */
private final AsyncTaskProgressService asyncTaskProgressService;
/**
* 创建异步任务进度查询接口。
*
* @param asyncTaskProgressService 通用 Redis 任务进度服务
*/
public AsyncTaskProgressController(AsyncTaskProgressService asyncTaskProgressService) {
this.asyncTaskProgressService = asyncTaskProgressService;
}
/**
* 查询当前用户的异步任务进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 上传或任务提交接口返回的不透明任务 ID
* @return 当前任务进度;任务不存在或已过期时返回失败响应
*/
@GetMapping("/progress")
@Operation(summary = "查询异步任务进度")
public R<AsyncTaskProgressVO> progress(@RequestParam String redisKey, @RequestParam String taskId) {
AsyncTaskProgressVO progress = asyncTaskProgressService.get(redisKey, taskId);
return progress == null ? R.failed("异步任务不存在或已过期") : R.ok(progress);
}
}
@@ -0,0 +1,16 @@
package com.bsd.pjcloud.common.asynctask.service;
/**
* 通用异步任务业务动作。
* <p>动作在后台事务中执行,抛出异常会回滚整个动作并把 Redis 任务标记为失败。</p>
*/
@FunctionalInterface
public interface AsyncTaskAction {
/**
* 执行具体异步业务。
*
* @param reporter 逐项进度报告器,业务完成一个计数项目后调用一次
*/
void execute(AsyncTaskProgressReporter reporter);
}
@@ -0,0 +1,7 @@
package com.bsd.pjcloud.common.asynctask.service;
/** 允许展示给任务发起人的受控错误;禁止传入 SQL、Token 或底层异常文本。 */
public class AsyncTaskException extends RuntimeException {
/** 使用业务可读提示终止任务,不暴露底层异常。 */
public AsyncTaskException(String message) { super(message); }
}
@@ -0,0 +1,34 @@
package com.bsd.pjcloud.common.asynctask.service;
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
/**
* 通用异步事务任务提交服务。
* <p>同步创建 Redis 任务后,把业务动作交给后台线程执行,并透传当前租户和登录用户上下文。</p>
*/
public interface AsyncTaskExecutorService {
/**
* 创建并提交一个异步事务任务。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数
* @param action 后台事务中执行的业务动作
* @return 等待执行的初始任务进度,调用方可立即把任务 ID 返回客户端
*/
AsyncTaskProgressVO submit(String redisKey, String taskName, int total, AsyncTaskAction action);
/**
* 创建并提交一个可在成功终态携带结果文件的异步事务任务。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数
* @param action 后台事务中执行并返回结果文件信息的业务动作
* @return 等待执行的初始任务进度,调用方可立即把任务 ID 返回客户端
*/
default AsyncTaskProgressVO submitWithResult(String redisKey, String taskName, int total, AsyncTaskResultAction action) {
throw new UnsupportedOperationException("当前异步任务执行器不支持返回任务结果");
}
}
@@ -0,0 +1,27 @@
package com.bsd.pjcloud.common.asynctask.service;
/**
* 异步任务逐项进度报告器。
* <p>业务每完成一个可计数项目后调用一次,由通用执行器递增 Redis 进度。</p>
*/
@FunctionalInterface
public interface AsyncTaskProgressReporter {
/**
* 在后台任务取得真实数据量后更新总项目数。
* <p>适用于查询本身耗时、必须先返回任务 ID 的导出场景;固定总数任务无需调用。</p>
*
* @param total 后台查询得到的真实总项目数,必须大于零
* @param message 总数更新后的进度说明;为空时使用通用说明
*/
default void updateTotal(int total, String message) {
throw new UnsupportedOperationException("当前异步任务进度报告器不支持动态更新总数");
}
/**
* 报告一个项目已处理完成。
*
* @param message 当前项目完成后的进度说明;为空时使用通用说明
*/
void advance(String message);
}
@@ -0,0 +1,91 @@
package com.bsd.pjcloud.common.asynctask.service;
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
/**
* 通用异步任务 Redis 进度服务。
* <p>业务 Key 按当前用户隔离,租户 ID 显式写入服务端任务键;调用方只提供业务命名空间和任务 ID。</p>
*/
public interface AsyncTaskProgressService {
/**
* 创建等待执行的异步任务并写入 Redis。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数,必须大于零
* @return 已写入 Redis 的初始任务进度
*/
AsyncTaskProgressVO create(String redisKey, String taskName, int total);
/**
* 将任务标记为执行中。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
*/
void start(String redisKey, String taskId);
/**
* 在任务开始后更新真实总项目数并重新计算完成百分比。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param total 后台查询得到的真实总项目数,必须大于零
* @param message 总数更新后的进度说明
*/
default void updateTotal(String redisKey, String taskId, int total, String message) {
throw new UnsupportedOperationException("当前异步任务进度服务不支持动态更新总数");
}
/**
* 更新任务执行进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param completed 当前已处理项目数
* @param message 当前进度说明
*/
void update(String redisKey, String taskId, int completed, String message);
/**
* 将任务标记为成功并把进度置为 100%。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 最终成功说明
*/
void success(String redisKey, String taskId, String message);
/**
* 将任务标记为成功,并在实现支持时保存业务动作返回的结果文件信息。
* <p>默认实现退化为原有成功状态写入,保证既有自定义进度服务无需同步修改;
* 需要向调用端返回附件的实现应覆写本方法。</p>
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 最终成功说明
* @param result 业务动作返回的可选结果文件信息
*/
default void success(String redisKey, String taskId, String message, AsyncTaskResult result) {
success(redisKey, taskId, message);
}
/**
* 将任务标记为失败并保留失败前执行进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 对调用端安全的失败说明
*/
void fail(String redisKey, String taskId, String message);
/**
* 查询当前用户创建的异步任务进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @return 任务进度;任务不存在、已过期或不属于当前用户时返回 {@code null}
*/
AsyncTaskProgressVO get(String redisKey, String taskId);
}
@@ -0,0 +1,40 @@
package com.bsd.pjcloud.common.asynctask.service;
/**
* 通用异步任务成功结果。
* <p>导入等无附件任务可返回 {@link #empty()},导出任务可返回下载地址和文件名称;结果只会在任务成功终态写入进度对象。</p>
*
* @param fileUrl 任务生成文件的下载地址;无结果文件时为 {@code null}
* @param fileName 下载时使用的业务文件名称;无结果文件时为 {@code null}
*/
public record AsyncTaskResult(String fileUrl, String fileName) {
/**
* 创建通用异步任务成功结果。
*
* @param fileUrl 任务生成文件的下载地址;无结果文件时为 {@code null}
* @param fileName 下载时使用的业务文件名称;无结果文件时为 {@code null}
*/
public AsyncTaskResult {
}
/**
* 创建不携带结果文件的成功结果。
*
* @return 无文件地址和名称的任务结果
*/
public static AsyncTaskResult empty() {
return new AsyncTaskResult(null, null);
}
/**
* 创建携带下载文件的成功结果。
*
* @param fileUrl 项目现有文件中心返回的文件下载地址
* @param fileName 下载时使用的业务文件名称
* @return 包含结果文件信息的任务结果
*/
public static AsyncTaskResult file(String fileUrl, String fileName) {
return new AsyncTaskResult(fileUrl, fileName);
}
}
@@ -0,0 +1,17 @@
package com.bsd.pjcloud.common.asynctask.service;
/**
* 携带成功结果的通用异步任务业务动作。
* <p>用于导出等需要在成功终态返回结果文件的任务;动作在后台事务中执行,抛出异常会回滚并标记任务失败。</p>
*/
@FunctionalInterface
public interface AsyncTaskResultAction {
/**
* 执行具体异步业务并返回成功结果。
*
* @param reporter 逐项进度报告器,业务完成一个计数项目后调用一次
* @return 任务成功终态需要返回调用端的可选结果文件信息
*/
AsyncTaskResult execute(AsyncTaskProgressReporter reporter) throws Exception;
}
@@ -0,0 +1,99 @@
package com.bsd.pjcloud.common.asynctask.service;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.bsd.pjcloud.common.excel.stream.BatchExcelExportTemplate;
import com.bsd.pjcloud.common.excel.stream.TemplateBatchWorkbook;
import jakarta.servlet.http.HttpServletResponse;
import org.apache.poi.xssf.usermodel.XSSFWorkbook;
import org.springframework.core.io.ClassPathResource;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Function;
/** 导出业务共用的分页、进度和完整文件响应适配,不改变业务字段转换。 */
public final class BatchExportSupport {
/** 一个查询批次的最大明细数,与模板流式行窗口一致。 */
public static final int BATCH_SIZE = 200;
/** XLSX 的附件响应类型。 */
public static final String CONTENT_TYPE = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private BatchExportSupport() { }
/** 按已有模板分页生成;仅在文件关闭后交给消费者上传或返回同步响应。 */
public static <T, R> R template(String template, Map<String, Object> values,
Function<Long, Page<T>> fetcher, Function<T, Map<String, Object>> converter,
AsyncTaskProgressReporter reporter, BatchExcelExportTemplate.FileConsumer<R> consumer,
Consumer<XSSFWorkbook> customizer) throws Exception {
Page<T> first = fetcher.apply(1L);
long total = first.getTotal();
if (total == 0) throw new AsyncTaskException("当前筛选条件没有可导出的数据");
if (reporter != null) {
reporter.updateTotal(BatchExcelExportTemplate.calculateProgressTotal(total, BATCH_SIZE, 2),
"共 " + total + " 条,正在分批生成 Excel");
}
try {
return TemplateBatchWorkbook.write(() -> new ClassPathResource(template).getInputStream(), total, BATCH_SIZE,
values, (current, size) -> current == 1 ? first.getRecords() : fetcher.apply(current).getRecords(),
converter, (completed, expected) -> {
if (reporter != null) reporter.advance("已写入 " + completed + " / " + expected + " 条");
}, file -> {
if (reporter != null) reporter.advance("文件已生成,正在保存导出结果");
return consumer.consume(file);
}, customizer);
} catch (IllegalArgumentException exception) {
// 模板行数校验的文案为受控提示,其他异常仍交给执行器统一处理。
if (exception.getMessage() != null && exception.getMessage().startsWith("导出明细数超过")) {
throw new AsyncTaskException(exception.getMessage() + ",请缩小筛选范围");
}
throw exception;
} catch (IllegalStateException exception) {
if (exception.getMessage() != null && exception.getMessage().startsWith("导出数据在执行期间发生变化")) {
throw new AsyncTaskException("导出期间数据发生变化,请重试");
}
throw exception;
}
}
/** 保持旧同步接口的附件语义;生成失败时尚未写出任何响应文件字节。 */
public static Void respond(Path file, String fileName, HttpServletResponse response) throws Exception {
response.setContentType(CONTENT_TYPE);
response.setHeader("Content-Disposition", "attachment;filename*=UTF-8''" +
URLEncoder.encode(fileName, StandardCharsets.UTF_8).replace("+", "%20"));
Files.copy(file, response.getOutputStream());
response.flushBuffer();
return null;
}
/** 延续旧模板的时间范围标题字段;缺失时间条件时占位符留空。 */
public static Map<String, Object> timeValues(LocalDateTime[] range) {
Map<String, Object> values = new HashMap<>();
if (range != null && range.length == 2 && range[0] != null && range[1] != null) {
DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
values.put("startTime", range[0].format(format));
values.put("endTime", range[1].format(format));
values.put("timeSpan", " 到 ");
}
return values;
}
/** 钱包日期区间沿用零点时间标题,不改变实际日期筛选。 */
public static Map<String, Object> timeValues(java.time.LocalDate[] range) {
if (range == null || range.length != 2) return new HashMap<>();
return timeValues(new LocalDateTime[]{range[0] == null ? null : range[0].atStartOfDay(), range[1] == null ? null : range[1].atStartOfDay()});
}
/** 创建受导出批大小约束的分页,后续页不重复执行 COUNT。 */
public static <T> Page<T> page(long current) {
Page<T> page = new Page<>(current, BATCH_SIZE, current == 1);
page.setMaxLimit((long) BATCH_SIZE);
return page;
}
}
@@ -0,0 +1,217 @@
package com.bsd.pjcloud.common.asynctask.service.impl;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskAction;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskException;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskExecutorService;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressReporter;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskResult;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskResultAction;
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
import com.bsd.pjcloud.common.security.util.NonWebTokenContextHolder;
import com.bsd.pjcloud.common.security.util.SecurityUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.task.TaskExecutor;
import org.springframework.security.core.Authentication;
import org.springframework.security.core.context.SecurityContext;
import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.security.core.context.SecurityContextHolderStrategy;
import org.springframework.transaction.support.TransactionTemplate;
import java.util.concurrent.atomic.AtomicInteger;
/**
* 通用异步事务任务提交服务默认实现。
* <p>实现会恢复提交线程的租户、安全和 Feign Token 上下文,并以单个 Spring 事务执行完整业务动作。</p>
*/
@Slf4j
public class DefaultAsyncTaskExecutorService implements AsyncTaskExecutorService {
/** 通用异步任务日志标签,用于统一检索任务提交和执行异常。 */
private static final String LOG_TAG = "通用异步任务";
/** 后台事务失败时返回调用端的固定说明,避免暴露 SQL 或远程调用等内部异常细节。 */
private static final String TASK_FAILED_MESSAGE = "导出任务执行失败,请重试";
/** 线程池拒绝任务时返回调用端的固定说明,提示稍后重新提交。 */
private static final String TASK_REJECTED_MESSAGE = "任务提交失败,请稍后重试";
/** 公共异步任务工作线程池,用于限制后台任务并发并与业务专用线程池隔离。 */
private final TaskExecutor asyncTaskWorkerExecutor;
/** Redis 任务进度服务,用于初始化、更新和查询任务状态。 */
private final AsyncTaskProgressService asyncTaskProgressService;
/** Spring 编程式事务模板,保证整个异步业务动作成功或整体回滚。 */
private final TransactionTemplate transactionTemplate;
/**
* 创建通用异步任务执行器。
*
* @param asyncTaskWorkerExecutor 公共异步任务专用工作线程池
* @param asyncTaskProgressService Redis 任务进度服务
* @param transactionTemplate Spring 编程式事务模板
*/
public DefaultAsyncTaskExecutorService(TaskExecutor asyncTaskWorkerExecutor,
AsyncTaskProgressService asyncTaskProgressService, TransactionTemplate transactionTemplate) {
this.asyncTaskWorkerExecutor = asyncTaskWorkerExecutor;
this.asyncTaskProgressService = asyncTaskProgressService;
this.transactionTemplate = transactionTemplate;
}
/**
* 创建并提交一个异步事务任务。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数
* @param action 后台事务中执行的业务动作
* @return 等待执行的初始任务进度
*/
@Override
public AsyncTaskProgressVO submit(String redisKey, String taskName, int total, AsyncTaskAction action) {
return submitWithResult(redisKey, taskName, total, reporter -> {
action.execute(reporter);
return AsyncTaskResult.empty();
});
}
/**
* 创建并提交一个可在成功终态携带结果文件的异步事务任务。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数
* @param action 后台事务中执行并返回结果文件信息的业务动作
* @return 等待执行的初始任务进度
*/
@Override
public AsyncTaskProgressVO submitWithResult(String redisKey, String taskName, int total, AsyncTaskResultAction action) {
AsyncTaskProgressVO progress = asyncTaskProgressService.create(redisKey, taskName, total);
Long tenantId = TenantContextHolder.getTenantId();
Authentication authentication = SecurityUtils.getAuthentication();
String token = SecurityUtils.getToken();
SecurityContextHolderStrategy strategy = SecurityContextHolder.getContextHolderStrategy();
SecurityContext securityContext = strategy.createEmptyContext();
securityContext.setAuthentication(authentication);
try {
asyncTaskWorkerExecutor.execute(() -> runTask(progress, tenantId, token, strategy, securityContext, action));
}
catch (RuntimeException exception) {
log.error("|{}|任务提交线程池失败,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
markFailedSafely(progress, TASK_REJECTED_MESSAGE);
throw new AsyncTaskException(TASK_REJECTED_MESSAGE);
}
return progress;
}
/**
* 在恢复租户、安全和 Feign Token 上下文后执行异步事务。
*
* @param progress 初始任务进度
* @param tenantId 请求提交时的租户 ID,恢复查询及任务键的租户隔离上下文
* @param token 请求提交时的 OAuth2 Token,可为空
* @param strategy 当前 Spring Security 上下文策略
* @param securityContext 请求提交时复制出的安全上下文
* @param action 后台事务中执行的业务动作
*/
private void runTask(AsyncTaskProgressVO progress, Long tenantId, String token, SecurityContextHolderStrategy strategy,
SecurityContext securityContext, AsyncTaskResultAction action) {
try {
TenantContextHolder.setTenantId(tenantId);
strategy.setContext(securityContext);
NonWebTokenContextHolder.setToken(token);
asyncTaskProgressService.start(progress.getRedisKey(), progress.getTaskId());
AtomicInteger completed = new AtomicInteger();
AsyncTaskResult result;
try {
AsyncTaskProgressReporter reporter = createProgressReporter(progress, completed);
result = transactionTemplate.execute(status -> {
try { return action.execute(reporter); }
catch (RuntimeException exception) { throw exception; }
catch (Exception exception) { throw new IllegalStateException("导出执行异常", exception); }
});
}
catch (Exception exception) {
log.error("|{}|任务事务执行失败,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
markFailedSafely(progress, exception instanceof AsyncTaskException ? exception.getMessage() : TASK_FAILED_MESSAGE);
return;
}
markSuccessSafely(progress, result);
}
catch (Exception exception) {
log.error("|{}|任务启动失败,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
markFailedSafely(progress, exception instanceof AsyncTaskException ? exception.getMessage() : TASK_FAILED_MESSAGE);
}
finally {
NonWebTokenContextHolder.clear();
strategy.clearContext();
TenantContextHolder.clear();
}
}
/**
* 创建同时支持逐项推进和动态总数更新的任务进度报告器。
*
* @param progress 当前任务的业务命名空间和任务 ID
* @param completed 当前任务线程内的已处理计数器
* @return 绑定当前任务 Redis 进度的报告器
*/
private AsyncTaskProgressReporter createProgressReporter(AsyncTaskProgressVO progress, AtomicInteger completed) {
return new AsyncTaskProgressReporter() {
/**
* 将后台查询得到的真实总数写入 Redis,并同步约束线程内已处理计数。
*
* @param total 真实总项目数,必须大于零
* @param message 总数更新后的进度说明
*/
@Override
public void updateTotal(int total, String message) {
completed.updateAndGet(current -> Math.min(current, total));
asyncTaskProgressService.updateTotal(progress.getRedisKey(), progress.getTaskId(), total, message);
}
/**
* 将已处理项目数递增一项并写入 Redis。
*
* @param message 当前项目完成后的进度说明
*/
@Override
public void advance(String message) {
int current = completed.incrementAndGet();
asyncTaskProgressService.update(progress.getRedisKey(), progress.getTaskId(), current, message);
}
};
}
/**
* 尝试把任务写为失败状态;Redis 异常只记录日志,不能覆盖原始业务异常或阻断线程上下文清理。
*
* @param progress 待更新的任务标识和业务命名空间
* @param message 对调用端安全的失败说明
*/
private void markFailedSafely(AsyncTaskProgressVO progress, String message) {
try {
asyncTaskProgressService.fail(progress.getRedisKey(), progress.getTaskId(), message);
}
catch (Exception exception) {
log.error("|{}|写入失败进度异常,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
}
}
/**
* 尝试把已提交事务的任务写为成功状态;Redis 写入失败只记录,不能把已经提交的数据误报为已回滚。
*
* @param progress 已完成数据库事务的任务标识和业务命名空间
* @param result 业务动作返回的可选结果文件信息
*/
private void markSuccessSafely(AsyncTaskProgressVO progress, AsyncTaskResult result) {
try {
asyncTaskProgressService.success(progress.getRedisKey(), progress.getTaskId(), "任务执行成功", result);
}
catch (Exception exception) {
log.error("|{}|写入成功进度异常,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
}
}
}
@@ -0,0 +1,262 @@
package com.bsd.pjcloud.common.asynctask.service.impl;
import com.bsd.pjcloud.common.asynctask.constant.AsyncTaskStatusEnum;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskResult;
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
import com.bsd.pjcloud.common.security.service.PjcloudUser;
import com.bsd.pjcloud.common.security.util.SecurityUtils;
import com.bsd.pjcloud.common.data.tenant.TenantContextHolder;
import org.springframework.data.redis.core.RedisTemplate;
import java.time.Duration;
import java.util.UUID;
import java.util.regex.Pattern;
/**
* 通用异步任务 Redis 进度服务默认实现。
* <p>该实现显式拼接租户、业务命名空间、登录用户和任务 ID,兼容项目当前普通字符串 Redis Key 序列化器。</p>
*/
public class DefaultAsyncTaskProgressService implements AsyncTaskProgressService {
/** 所有通用异步任务 Redis Key 的固定业务命名空间。 */
private static final String REDIS_KEY_PREFIX = "async:task:progress";
/** 任务状态的 Redis 保留时间,供页面刷新或短时重连后继续查询。 */
private static final Duration TASK_TTL = Duration.ofHours(24);
/** 业务 Redis Key 允许的字符和长度,防止调用方构造不可控的 Redis 路径。 */
private static final Pattern REDIS_KEY_PATTERN = Pattern.compile("^[a-zA-Z0-9:_-]{1,80}$");
/** 不透明任务 ID 的固定格式,与本服务生成的 32 位 UUID 一致。 */
private static final Pattern TASK_ID_PATTERN = Pattern.compile("^[a-fA-F0-9]{32}$");
/** Redis 操作模板,使用项目统一 JSON 序列化配置保存进度对象。 */
private final RedisTemplate<String, Object> redisTemplate;
/**
* 创建通用 Redis 进度服务。
*
* @param redisTemplate 项目统一配置的 Redis 操作模板
*/
public DefaultAsyncTaskProgressService(RedisTemplate<String, Object> redisTemplate) {
this.redisTemplate = redisTemplate;
}
/**
* 创建等待执行的异步任务并写入 Redis。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskName 任务展示名称
* @param total 需要处理的总项目数,必须大于零
* @return 已写入 Redis 的初始任务进度
*/
@Override
public AsyncTaskProgressVO create(String redisKey, String taskName, int total) {
validateRedisKey(redisKey);
if (total <= 0) {
throw new IllegalArgumentException("异步任务总数必须大于0");
}
AsyncTaskProgressVO progress = new AsyncTaskProgressVO();
progress.setTaskId(UUID.randomUUID().toString().replace("-", ""));
progress.setRedisKey(redisKey);
progress.setTaskName(taskName);
progress.setStatus(AsyncTaskStatusEnum.PENDING.getStatus());
progress.setTotal(total);
progress.setCompleted(0);
progress.setPercentage(0);
progress.setMessage(AsyncTaskStatusEnum.PENDING.getDescription());
progress.setFinished(false);
progress.setSuccess(false);
save(progress);
return progress;
}
/**
* 将任务标记为执行中。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
*/
@Override
public void start(String redisKey, String taskId) {
AsyncTaskProgressVO progress = requireProgress(redisKey, taskId);
progress.setStatus(AsyncTaskStatusEnum.RUNNING.getStatus());
progress.setMessage(AsyncTaskStatusEnum.RUNNING.getDescription());
save(progress);
}
/**
* 在任务开始后更新真实总项目数并重新计算完成百分比。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param total 后台查询得到的真实总项目数,必须大于零
* @param message 总数更新后的进度说明
*/
@Override
public void updateTotal(String redisKey, String taskId, int total, String message) {
if (total <= 0) {
throw new IllegalArgumentException("异步任务总数必须大于0");
}
AsyncTaskProgressVO progress = requireProgress(redisKey, taskId);
int safeCompleted = Math.max(0, Math.min(progress.getCompleted(), total));
progress.setStatus(AsyncTaskStatusEnum.RUNNING.getStatus());
progress.setTotal(total);
progress.setCompleted(safeCompleted);
progress.setPercentage(calculatePercentage(safeCompleted, total));
progress.setMessage(message == null || message.isBlank() ? "正在处理" : message);
save(progress);
}
/**
* 更新任务执行进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param completed 当前已处理项目数
* @param message 当前进度说明
*/
@Override
public void update(String redisKey, String taskId, int completed, String message) {
AsyncTaskProgressVO progress = requireProgress(redisKey, taskId);
int safeCompleted = Math.max(0, Math.min(completed, progress.getTotal()));
progress.setStatus(AsyncTaskStatusEnum.RUNNING.getStatus());
progress.setCompleted(safeCompleted);
progress.setPercentage(calculatePercentage(safeCompleted, progress.getTotal()));
progress.setMessage(message == null || message.isBlank() ? "正在处理" : message);
save(progress);
}
/**
* 将任务标记为成功并把进度置为 100%。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 最终成功说明
*/
@Override
public void success(String redisKey, String taskId, String message) {
success(redisKey, taskId, message, null);
}
/**
* 将任务标记为成功、把进度置为 100%,并保存可选的结果文件信息。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 最终成功说明
* @param result 业务动作返回的可选结果文件信息
*/
@Override
public void success(String redisKey, String taskId, String message, AsyncTaskResult result) {
AsyncTaskProgressVO progress = requireProgress(redisKey, taskId);
progress.setStatus(AsyncTaskStatusEnum.SUCCESS.getStatus());
progress.setCompleted(progress.getTotal());
progress.setPercentage(100);
progress.setMessage(message == null || message.isBlank() ? AsyncTaskStatusEnum.SUCCESS.getDescription() : message);
progress.setFinished(true);
progress.setSuccess(true);
if (result != null) {
progress.setResultFileUrl(result.fileUrl());
progress.setResultFileName(result.fileName());
}
save(progress);
}
/**
* 将任务标记为失败并保留失败前执行进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @param message 对调用端安全的失败说明
*/
@Override
public void fail(String redisKey, String taskId, String message) {
AsyncTaskProgressVO progress = requireProgress(redisKey, taskId);
progress.setStatus(AsyncTaskStatusEnum.FAILED.getStatus());
progress.setMessage(message == null || message.isBlank() ? AsyncTaskStatusEnum.FAILED.getDescription() : message);
progress.setFinished(true);
progress.setSuccess(false);
save(progress);
}
/**
* 查询当前用户创建的异步任务进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @return 任务进度;任务不存在、已过期或不属于当前用户时返回 {@code null}
*/
@Override
public AsyncTaskProgressVO get(String redisKey, String taskId) {
validateRedisKey(redisKey);
Object value = redisTemplate.opsForValue().get(buildRedisKey(redisKey, taskId));
return value instanceof AsyncTaskProgressVO progress ? progress : null;
}
/**
* 获取必须存在的任务进度。
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @return 当前任务进度
*/
private AsyncTaskProgressVO requireProgress(String redisKey, String taskId) {
AsyncTaskProgressVO progress = get(redisKey, taskId);
if (progress == null) {
throw new IllegalStateException("异步任务不存在或已过期");
}
return progress;
}
/**
* 保存任务进度并刷新统一过期时间。
*
* @param progress 待保存的完整进度对象
*/
private void save(AsyncTaskProgressVO progress) {
redisTemplate.opsForValue().set(buildRedisKey(progress.getRedisKey(), progress.getTaskId()), progress, TASK_TTL);
}
/**
* 构建按当前用户隔离的业务 Redis Key。
* <p>租户 ID 显式进入任务键,不依赖 Redis 序列化器隐式隔离。</p>
*
* @param redisKey 业务指定的 Redis Key 命名空间
* @param taskId 不透明任务 ID
* @return 仅服务端使用的业务 Redis Key,结构为固定前缀、租户 ID、业务命名空间、用户 ID 和任务 ID
*/
private String buildRedisKey(String redisKey, String taskId) {
PjcloudUser currentUser = SecurityUtils.getUser();
if (currentUser == null || currentUser.getId() == null) {
throw new IllegalStateException("缺少异步任务用户上下文");
}
if (taskId == null || !TASK_ID_PATTERN.matcher(taskId).matches()) {
throw new IllegalArgumentException("异步任务ID格式不正确");
}
return String.join(":", REDIS_KEY_PREFIX, String.valueOf(TenantContextHolder.getTenantId()), redisKey, String.valueOf(currentUser.getId()), taskId);
}
/**
* 校验业务 Redis Key 命名空间,限制长度与字符范围。
*
* @param redisKey 业务指定的 Redis Key 命名空间
*/
private void validateRedisKey(String redisKey) {
if (redisKey == null || !REDIS_KEY_PATTERN.matcher(redisKey).matches()) {
throw new IllegalArgumentException("异步任务Redis Key格式不正确");
}
}
/**
* 计算取整后的完成百分比。
*
* @param completed 已完成项目数
* @param total 总项目数
* @return 0 至 99 的整数百分比;只有 success 终态可写入 100
*/
private int calculatePercentage(int completed, int total) {
return total <= 0 ? 0 : (int) Math.min(99L, (long) completed * 100 / total);
}
}
@@ -0,0 +1,66 @@
package com.bsd.pjcloud.common.asynctask.vo;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.Data;
import java.io.Serializable;
/**
* 通用异步任务进度展示对象。
* <p>对象存入 Redis 并通过通用进度接口返回,可用于导入、导出等后台任务。</p>
*/
@Data
@Schema(description = "通用异步任务进度")
public class AsyncTaskProgressVO implements Serializable {
/** 序列化版本号,用于 Redis 中任务状态的兼容反序列化。 */
private static final long serialVersionUID = 1L;
/** 调用端轮询使用的不透明任务 ID,不包含租户或用户信息。 */
@Schema(description = "任务ID")
private String taskId;
/** 业务指定的 Redis Key 命名空间,用于区分不同异步任务场景。 */
@Schema(description = "Redis Key命名空间")
private String redisKey;
/** 任务展示名称,例如后台用户导入。 */
@Schema(description = "任务名称")
private String taskName;
/** 固定任务状态编码:PENDING、RUNNING、SUCCESS、FAILED。 */
@Schema(description = "任务状态")
private String status;
/** 任务需要处理的总项目数。 */
@Schema(description = "总项目数")
private Integer total;
/** 当前执行到的项目数;任务整体失败时仅表示回滚或失败前的执行位置,不代表最终成功写入数量。 */
@Schema(description = "已处理项目数")
private Integer completed;
/** 任务完成百分比,取值范围为 0 至 100。 */
@Schema(description = "完成百分比")
private Integer percentage;
/** 当前进度或最终结果说明。 */
@Schema(description = "任务消息")
private String message;
/** 是否已进入成功或失败终态,供调用端决定是否停止轮询。 */
@Schema(description = "是否完成")
private Boolean finished;
/** 终态是否成功;任务未结束时固定为 {@code false}。 */
@Schema(description = "是否成功")
private Boolean success;
/** 任务成功后生成文件的下载地址;导入等无附件任务为空。 */
@Schema(description = "任务结果文件下载地址")
private String resultFileUrl;
/** 任务成功后生成文件的下载名称;导入等无附件任务为空。 */
@Schema(description = "任务结果文件名称")
private String resultFileName;
}
@@ -0,0 +1,2 @@
com.bsd.pjcloud.common.asynctask.config.AsyncTaskAutoConfiguration
com.bsd.pjcloud.common.asynctask.config.AsyncTaskWebAutoConfiguration