feat:增加老师额外的用户字段。
feat:增加了老师的导入,和导入的进度条。 feat:app分配次退款。 feat:上传人脸不需要审核。 feat:部门管理员导入用户不需要审核。
This commit is contained in:
1 parent
07028116c1
commit
5a7f8dcfbb
62 files changed
+3884
-760
No files matched your search
+95
@@ -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);
|
||||
}
|
||||
}
|
||||
+33
@@ -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);
|
||||
}
|
||||
}
|
||||
+40
@@ -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;
|
||||
}
|
||||
}
|
||||
+50
@@ -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 按登录用户隔离,租户前缀由项目 Redis 基础设施统一处理,只能查询本人创建且尚未过期的任务。</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);
|
||||
}
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
package com.bsd.pjcloud.common.asynctask.service;
|
||||
|
||||
/**
|
||||
* 通用异步任务业务动作。
|
||||
* <p>动作在后台事务中执行,抛出异常会回滚整个动作并把 Redis 任务标记为失败。</p>
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface AsyncTaskAction {
|
||||
|
||||
/**
|
||||
* 执行具体异步业务。
|
||||
*
|
||||
* @param reporter 逐项进度报告器,业务完成一个计数项目后调用一次
|
||||
*/
|
||||
void execute(AsyncTaskProgressReporter reporter);
|
||||
}
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
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);
|
||||
}
|
||||
+16
@@ -0,0 +1,16 @@
|
||||
package com.bsd.pjcloud.common.asynctask.service;
|
||||
|
||||
/**
|
||||
* 异步任务逐项进度报告器。
|
||||
* <p>业务每完成一个可计数项目后调用一次,由通用执行器递增 Redis 进度。</p>
|
||||
*/
|
||||
@FunctionalInterface
|
||||
public interface AsyncTaskProgressReporter {
|
||||
|
||||
/**
|
||||
* 报告一个项目已处理完成。
|
||||
*
|
||||
* @param message 当前项目完成后的进度说明;为空时使用通用说明
|
||||
*/
|
||||
void advance(String message);
|
||||
}
|
||||
+65
@@ -0,0 +1,65 @@
|
||||
package com.bsd.pjcloud.common.asynctask.service;
|
||||
|
||||
import com.bsd.pjcloud.common.asynctask.vo.AsyncTaskProgressVO;
|
||||
|
||||
/**
|
||||
* 通用异步任务 Redis 进度服务。
|
||||
* <p>业务 Key 按当前用户隔离,租户前缀由项目 Redis 基础设施统一处理;调用方只提供业务命名空间和任务 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 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);
|
||||
|
||||
/**
|
||||
* 将任务标记为失败并保留失败前执行进度。
|
||||
*
|
||||
* @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);
|
||||
}
|
||||
+158
@@ -0,0 +1,158 @@
|
||||
package com.bsd.pjcloud.common.asynctask.service.impl;
|
||||
|
||||
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskAction;
|
||||
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskExecutorService;
|
||||
import com.bsd.pjcloud.common.asynctask.service.AsyncTaskProgressService;
|
||||
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) {
|
||||
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 exception;
|
||||
}
|
||||
return progress;
|
||||
}
|
||||
|
||||
/**
|
||||
* 在恢复租户、安全和 Feign Token 上下文后执行异步事务。
|
||||
*
|
||||
* @param progress 初始任务进度
|
||||
* @param tenantId 请求提交时的租户 ID,仅用于恢复基础设施上下文,不参与业务 Redis Key 拼接
|
||||
* @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, AsyncTaskAction action) {
|
||||
try {
|
||||
TenantContextHolder.setTenantId(tenantId);
|
||||
strategy.setContext(securityContext);
|
||||
NonWebTokenContextHolder.setToken(token);
|
||||
asyncTaskProgressService.start(progress.getRedisKey(), progress.getTaskId());
|
||||
AtomicInteger completed = new AtomicInteger();
|
||||
try {
|
||||
transactionTemplate.executeWithoutResult(status -> action.execute(message -> {
|
||||
int current = completed.incrementAndGet();
|
||||
asyncTaskProgressService.update(progress.getRedisKey(), progress.getTaskId(), current, message);
|
||||
}));
|
||||
}
|
||||
catch (Exception exception) {
|
||||
log.error("|{}|任务事务执行失败,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
|
||||
markFailedSafely(progress, TASK_FAILED_MESSAGE);
|
||||
return;
|
||||
}
|
||||
markSuccessSafely(progress);
|
||||
}
|
||||
catch (Exception exception) {
|
||||
log.error("|{}|任务启动失败,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
|
||||
markFailedSafely(progress, TASK_FAILED_MESSAGE);
|
||||
}
|
||||
finally {
|
||||
NonWebTokenContextHolder.clear();
|
||||
strategy.clearContext();
|
||||
TenantContextHolder.clear();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 尝试把任务写为失败状态;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 已完成数据库事务的任务标识和业务命名空间
|
||||
*/
|
||||
private void markSuccessSafely(AsyncTaskProgressVO progress) {
|
||||
try {
|
||||
asyncTaskProgressService.success(progress.getRedisKey(), progress.getTaskId(), "任务执行成功");
|
||||
}
|
||||
catch (Exception exception) {
|
||||
log.error("|{}|写入成功进度异常,taskId:{},taskName:{}", LOG_TAG, progress.getTaskId(), progress.getTaskName(), exception);
|
||||
}
|
||||
}
|
||||
}
|
||||
+220
@@ -0,0 +1,220 @@
|
||||
package com.bsd.pjcloud.common.asynctask.service.impl;
|
||||
|
||||
import com.bsd.pjcloud.common.asynctask.constant.AsyncTaskStatusEnum;
|
||||
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 org.springframework.data.redis.core.RedisTemplate;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.UUID;
|
||||
import java.util.regex.Pattern;
|
||||
|
||||
/**
|
||||
* 通用异步任务 Redis 进度服务默认实现。
|
||||
* <p>该实现只拼接业务命名空间、登录用户和任务 ID,租户维度由项目 Redis 基础设施根据当前上下文统一处理。</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 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) {
|
||||
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);
|
||||
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>租户前缀由项目统一 Redis 基础设施自动处理,此处不得重复拼接租户 ID。</p>
|
||||
*
|
||||
* @param redisKey 业务指定的 Redis Key 命名空间
|
||||
* @param taskId 不透明任务 ID
|
||||
* @return 仅服务端使用的业务 Redis Key,结构为固定前缀、业务命名空间、用户 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, 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 至 100 的整数百分比
|
||||
*/
|
||||
private int calculatePercentage(int completed, int total) {
|
||||
return total <= 0 ? 0 : Math.min(100, completed * 100 / total);
|
||||
}
|
||||
}
|
||||
+58
@@ -0,0 +1,58 @@
|
||||
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;
|
||||
}
|
||||
+2
@@ -0,0 +1,2 @@
|
||||
com.bsd.pjcloud.common.asynctask.config.AsyncTaskAutoConfiguration
|
||||
com.bsd.pjcloud.common.asynctask.config.AsyncTaskWebAutoConfiguration
|
||||
Reference in new issue
Block a user