refactor: 使用事件解耦业务副作用并优化虚拟线程并发

- 新增登录成功事件，解耦在线用户缓存、登录日志和最近登录信息更新 - 新增在线用户清理事件，角色授权变更后异步清理受影响用户会话 - 新增 OSS 配置变更事件，统一处理配置初始化、删除和默认配置切换后的缓存刷新 - 新增工作流抄送、待办消息、结果消息事件，拆分全局监听器中的副作用逻辑 - 扩展 ThreadUtils 支持带返回值的虚拟线程批量执行 - 在线用户监控批量读取 Redis 会话信息时使用虚拟线程并发处理 - OSS 多文件查询、URL 解析和 DTO 转换使用虚拟线程并发处理 - 工作流办理人回显和用户解析按类型并发查询 - 工作流多渠道消息发送改为并发执行 - WebSocket 与 SSE 广播消息按用户并发发送
2026-05-18 17:22:13 +08:00
parent 4383e8f1fc
commit ae685cc0d5
26 changed files with 587 additions and 192 deletions
@@ -4,8 +4,10 @@ import lombok.AccessLevel;
 import lombok.NoArgsConstructor;

 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.*;
+import java.util.function.Supplier;

 /**
 * 线程工具
@@ -47,4 +49,43 @@ public class ThreadUtils {
        virtualInvokeAll(runnableList);
    }

+    /**
+     * 批量执行有返回值的任务
+     *
+     * @param supplierList 任务列表
+     * @param <T>          返回值类型
+     * @return 按提交顺序返回的任务结果
+     */
+    @SafeVarargs
+    public static <T> List<T> virtualSubmitAll(Supplier<T>... supplierList) {
+        return virtualSubmitAll(List.of(supplierList));
+    }
+
+    /**
+     * 批量执行有返回值的任务
+     *
+     * @param supplierList 任务列表
+     * @param <T>          返回值类型
+     * @return 按提交顺序返回的任务结果
+     */
+    public static <T> List<T> virtualSubmitAll(Collection<? extends Supplier<T>> supplierList) {
+        List<Future<T>> futureList = new ArrayList<>();
+        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
+            for (Supplier<T> supplier : supplierList) {
+                futureList.add(executor.submit(supplier::get));
+            }
+            List<T> resultList = new ArrayList<>(futureList.size());
+            for (Future<T> future : futureList) {
+                resultList.add(future.get());
+            }
+            return resultList;
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new RuntimeException("线程执行被中断", e);
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause() == null ? e : e.getCause();
+            throw new RuntimeException("线程执行异常：" + cause.getMessage(), cause);
+        }
+    }
+
 }
@@ -3,6 +3,7 @@ package org.dromara.common.push.core;
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.map.MapUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.utils.ThreadUtils;
 import org.dromara.common.json.utils.JsonUtils;
 import org.dromara.common.push.constant.MessageConstants;
 import org.dromara.common.push.dto.PushDTO;
@@ -227,9 +228,11 @@ public class SseEmitterSessionManager implements PushSessionManager {
     * @param message 要发送的消息内容
     */
    public void sendMessage(String message) {
-        for (Long userId : USER_TOKEN_EMITTERS.keySet()) {
-            sendMessage(userId, message);
-        }
+        List<Long> userIds = new ArrayList<>(USER_TOKEN_EMITTERS.keySet());
+        Runnable[] sendTasks = userIds.stream()
+            .map(userId -> (Runnable) () -> sendMessage(userId, message))
+            .toArray(Runnable[]::new);
+        ThreadUtils.virtualInvokeAll(sendTasks);
    }

    /**
@@ -3,6 +3,7 @@ package org.dromara.common.push.core;
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.map.MapUtil;
 import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.utils.ThreadUtils;
 import org.dromara.common.json.utils.JsonUtils;
 import org.dromara.common.push.dto.PushDTO;
 import org.dromara.common.push.dto.PushPayloadDTO;
@@ -181,7 +182,11 @@ public class WebSocketSessionManager implements PushSessionManager {
        if (payload == null) {
            return;
        }
-        USER_TOKEN_SESSIONS.keySet().forEach(userId -> sendMessage(userId, payload));
+        List<Long> userIds = new ArrayList<>(USER_TOKEN_SESSIONS.keySet());
+        Runnable[] sendTasks = userIds.stream()
+            .map(userId -> (Runnable) () -> sendMessage(userId, payload))
+            .toArray(Runnable[]::new);
+        ThreadUtils.virtualInvokeAll(sendTasks);
    }

    /**
@@ -33,7 +33,7 @@
                </exclusion>
            </exclusions>
        </dependency>
-        <!-- Boot 4 下使用 jetty 替代 undertow -->
+        <!-- web 容器使用 jetty -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jetty</artifactId>
@@ -3,7 +3,6 @@ package org.dromara.common.web.config;
 import cn.hutool.core.date.DateTime;
 import cn.hutool.core.date.DateUtil;
 import org.dromara.common.core.utils.ObjectUtils;
-import org.dromara.common.core.utils.StringUtils;
 import org.dromara.common.json.enhance.JsonValueEnhancer;
 import org.dromara.common.web.advice.ResponseEnhancementAdvice;
 import org.dromara.common.web.handler.GlobalExceptionHandler;
@@ -1,43 +0,0 @@
-//package org.dromara.common.web.config;
-//
-//import io.undertow.server.DefaultByteBufferPool;
-//import io.undertow.server.handlers.DisallowedMethodsHandler;
-//import io.undertow.util.HttpString;
-//import io.undertow.websockets.jsr.WebSocketDeploymentInfo;
-//import org.dromara.common.core.utils.SpringUtils;
-//import org.springframework.boot.autoconfigure.AutoConfiguration;
-//import org.springframework.boot.web.embedded.undertow.UndertowServletWebServerFactory;
-//import org.springframework.boot.web.server.WebServerFactoryCustomizer;
-//import org.springframework.core.task.VirtualThreadTaskExecutor;
-//
-///**
-// * Boot 4 迁移后默认改用 Jetty，这里暂时保留 Undertow 配置实现以便后续回切时参考。
-// */
-//@AutoConfiguration
-//public class UndertowConfig implements WebServerFactoryCustomizer<UndertowServletWebServerFactory> {
-//
-//    @Override
-//    public void customize(UndertowServletWebServerFactory factory) {
-//        factory.addDeploymentInfoCustomizers(deploymentInfo -> {
-//            WebSocketDeploymentInfo webSocketDeploymentInfo = new WebSocketDeploymentInfo();
-//            webSocketDeploymentInfo.setBuffers(new DefaultByteBufferPool(true, 1024));
-//            deploymentInfo.addServletContextAttribute("io.undertow.websockets.jsr.WebSocketDeploymentInfo", webSocketDeploymentInfo);
-//
-//            if (SpringUtils.isVirtual()) {
-//                VirtualThreadTaskExecutor executor = new VirtualThreadTaskExecutor("undertow-");
-//                deploymentInfo.setExecutor(executor);
-//                deploymentInfo.setAsyncExecutor(executor);
-//            }
-//
-//            deploymentInfo.addInitialHandlerChainWrapper(handler -> {
-//                HttpString[] disallowedHttpMethods = {
-//                    HttpString.tryFromString("CONNECT"),
-//                    HttpString.tryFromString("TRACE"),
-//                    HttpString.tryFromString("TRACK")
-//                };
-//                return new DisallowedMethodsHandler(handler, disallowedHttpMethods);
-//            });
-//        });
-//    }
-//
-//}
@@ -11,7 +11,6 @@ import org.dromara.common.core.domain.R;
 import org.dromara.common.core.exception.ServiceException;
 import org.dromara.common.core.exception.SseException;
 import org.dromara.common.core.exception.base.BaseException;
-import org.dromara.common.core.utils.SpringUtils;
 import org.dromara.common.core.utils.StreamUtils;
 import org.dromara.common.json.utils.JsonUtils;
 import org.springframework.boot.json.JsonParseException;
@@ -212,7 +211,7 @@ public class GlobalExceptionHandler {
    public R<Void> handleJsonParseException(JsonParseException e, HttpServletRequest request) {
        String requestURI = request.getRequestURI();
        log.error("请求地址'{}' 发生 JSON 解析异常: {}", requestURI, e.getMessage());
-        return R.fail(HttpStatus.HTTP_BAD_REQUEST, "请求数据格式错误（JSON 解析失败）：" + e.getMessage());
+        return R.fail(HttpStatus.HTTP_BAD_REQUEST, "请求数据格式错误");
    }

    /**
@@ -221,7 +220,7 @@ public class GlobalExceptionHandler {
    @ExceptionHandler(HttpMessageNotReadableException.class)
    public R<Void> handleHttpMessageNotReadableException(HttpMessageNotReadableException e, HttpServletRequest request) {
        log.error("请求地址'{}', 参数解析失败: {}", request.getRequestURI(), e.getMessage());
-        return R.fail(HttpStatus.HTTP_BAD_REQUEST, "请求参数格式错误：" + e.getMostSpecificCause().getMessage());
+        return R.fail(HttpStatus.HTTP_BAD_REQUEST, "请求参数格式错误");
    }

    /**