xiaozhi/xiaozhi-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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

Browse Source 
- 新增登录成功事件，解耦在线用户缓存、登录日志和最近登录信息更新
  - 新增在线用户清理事件，角色授权变更后异步清理受影响用户会话
  - 新增 OSS 配置变更事件，统一处理配置初始化、删除和默认配置切换后的缓存刷新
  - 新增工作流抄送、待办消息、结果消息事件，拆分全局监听器中的副作用逻辑
  - 扩展 ThreadUtils 支持带返回值的虚拟线程批量执行
  - 在线用户监控批量读取 Redis 会话信息时使用虚拟线程并发处理
  - OSS 多文件查询、URL 解析和 DTO 转换使用虚拟线程并发处理
  - 工作流办理人回显和用户解析按类型并发查询
  - 工作流多渠道消息发送改为并发执行
  - WebSocket 与 SSE 广播消息按用户并发发送
This commit is contained in:
疯狂的狮子Li
2026-05-16 18:44:24 +08:00
parent f171ac03c4
commit 84afd6e6c0
23 changed files with 579 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/event/WorkflowCopyEvent.java
+23
View File
@@ -0,0 +1,23 @@
package org.dromara.workflow.event;
import cn.hutool.core.collection.CollUtil;
import org.dromara.warm.flow.core.entity.Task;
import org.dromara.workflow.domain.bo.FlowCopyBo;
import java.util.List;
/**
* 工作流抄送事件。
*
* @param task 当前任务
* @param flowCopyList 抄送人列表
*/
public record WorkflowCopyEvent(Task task, List<FlowCopyBo> flowCopyList) {
public WorkflowCopyEvent {
if (CollUtil.isNotEmpty(flowCopyList)) {
flowCopyList = List.copyOf(flowCopyList);
}
}
}
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/event/WorkflowResultMessageEvent.java
+23
View File
@@ -0,0 +1,23 @@
package org.dromara.workflow.event;
import cn.hutool.core.collection.CollUtil;
import java.util.List;
/**
* 工作流结果消息事件。
*
* @param flowName 流程名称
* @param status 流程状态
* @param createBy 发起人用户 ID
* @param messageType 消息类型
*/
public record WorkflowResultMessageEvent(String flowName, String status, String createBy, List<String> messageType) {
public WorkflowResultMessageEvent {
if (CollUtil.isNotEmpty(messageType)) {
messageType = List.copyOf(messageType);
}
}
}
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/event/WorkflowTaskMessageEvent.java
+23
View File
@@ -0,0 +1,23 @@
package org.dromara.workflow.event;
import cn.hutool.core.collection.CollUtil;
import java.util.List;
/**
* 工作流待办消息事件。
*
* @param flowName 流程名称
* @param instanceId 流程实例 ID
* @param messageType 消息类型
* @param notice 通知内容
*/
public record WorkflowTaskMessageEvent(String flowName, Long instanceId, List<String> messageType, String notice) {
public WorkflowTaskMessageEvent {
if (CollUtil.isNotEmpty(messageType)) {
messageType = List.copyOf(messageType);
}
}
}
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/listener/WorkflowGlobalListener.java
+8 -18
View File
@@ -9,10 +9,10 @@ import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.enums.BusinessStatusEnum;
import org.dromara.common.core.utils.SpringUtils;
import org.dromara.common.core.utils.StreamUtils;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.system.api.UserService;
import org.dromara.system.api.domain.UserDTO;
import org.dromara.warm.flow.core.FlowEngine;
import org.dromara.warm.flow.core.dto.FlowParams;
import org.dromara.warm.flow.core.entity.Definition;
@@ -22,10 +22,12 @@ import org.dromara.warm.flow.core.listener.GlobalListener;
import org.dromara.warm.flow.core.listener.ListenerVariable;
import org.dromara.workflow.common.ConditionalOnEnable;
import org.dromara.workflow.common.constant.FlowConstant;
import org.dromara.workflow.common.enums.MessageTypeEnum;
import org.dromara.workflow.common.enums.TaskStatusEnum;
import org.dromara.workflow.domain.bo.FlowCopyBo;
import org.dromara.workflow.domain.vo.NodeExtVo;
import org.dromara.workflow.event.WorkflowCopyEvent;
import org.dromara.workflow.event.WorkflowResultMessageEvent;
import org.dromara.workflow.event.WorkflowTaskMessageEvent;
import org.dromara.workflow.handler.FlowProcessEventHandler;
import org.dromara.workflow.service.IFlwCommonService;
import org.dromara.workflow.service.IFlwInstanceService;
@@ -232,7 +234,7 @@ public class WorkflowGlobalListener implements GlobalListener {
List<FlowCopyBo> flowCopyList = MapUtil.get(variable, FlowConstant.FLOW_COPY_LIST, new TypeReference<>() {
});
// 添加抄送人
flwTaskService.setCopy(task, flowCopyList);
SpringUtils.context().publishEvent(new WorkflowCopyEvent(task, flowCopyList));
}
if (variable.containsKey(FlowConstant.MESSAGE_TYPE)) {
List<String> messageType = MapUtil.get(variable, FlowConstant.MESSAGE_TYPE, new TypeReference<>() {
@@ -240,7 +242,7 @@ public class WorkflowGlobalListener implements GlobalListener {
String notice = MapUtil.getStr(variable, FlowConstant.MESSAGE_NOTICE);
// 退回到申请人时只保留“已退回”结果消息，避免再追加一条“新的待办”形成重复提醒。
if (shouldSendTaskMessage(flowParams, definition, nextTasks)) {
flwCommonService.sendMessage(definition.getFlowName(), instance.getId(), messageType, notice);
SpringUtils.context().publishEvent(new WorkflowTaskMessageEvent(definition.getFlowName(), instance.getId(), messageType, notice));
}
}
FlowEngine.insService().removeVariables(instance.getId(),
@@ -270,25 +272,13 @@ public class WorkflowGlobalListener implements GlobalListener {
if (StringUtils.isBlank(instance.getCreateBy())) {
return;
}
BusinessStatusEnum statusEnum = BusinessStatusEnum.getByStatus(status);
if (statusEnum == null) {
return;
}
Long createBy = Convert.toLong(instance.getCreateBy(), null);
if (createBy == null) {
return;
}
UserDTO initiator = userService.selectById(createBy);
if (initiator == null || initiator.getUserId() == null) {
return;
}
// 已完成、已退回这类结果消息只发给发起人，不再混入处理人待办消息。
List<String> messageType = Collections.singletonList(MessageTypeEnum.SYSTEM_MESSAGE.getCode());
List<String> messageType = null;
if (MapUtil.isNotEmpty(variable) && variable.containsKey(FlowConstant.MESSAGE_TYPE)) {
messageType = MapUtil.get(variable, FlowConstant.MESSAGE_TYPE, new TypeReference<>() {
});
}
flwCommonService.sendResultMessage(definition.getFlowName(), statusEnum, messageType, Collections.singletonList(initiator));
SpringUtils.context().publishEvent(new WorkflowResultMessageEvent(definition.getFlowName(), status, instance.getCreateBy(), messageType));
}
/**
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/listener/WorkflowSideEffectListener.java
+85
View File
@@ -0,0 +1,85 @@
package org.dromara.workflow.listener;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import lombok.RequiredArgsConstructor;
import org.dromara.common.core.enums.BusinessStatusEnum;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.system.api.UserService;
import org.dromara.system.api.domain.UserDTO;
import org.dromara.workflow.common.ConditionalOnEnable;
import org.dromara.workflow.common.enums.MessageTypeEnum;
import org.dromara.workflow.event.WorkflowCopyEvent;
import org.dromara.workflow.event.WorkflowResultMessageEvent;
import org.dromara.workflow.event.WorkflowTaskMessageEvent;
import org.dromara.workflow.service.IFlwCommonService;
import org.dromara.workflow.service.IFlwTaskService;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Component;
import java.util.Collections;
import java.util.List;
/**
* 工作流副作用事件监听器。
*
* @author may
*/
@ConditionalOnEnable
@Component
@RequiredArgsConstructor
public class WorkflowSideEffectListener {
private final IFlwTaskService flwTaskService;
private final IFlwCommonService flwCommonService;
private final UserService userService;
/**
* 保存工作流抄送记录。
*
* @param event 工作流抄送事件
*/
@EventListener
public void handleCopy(WorkflowCopyEvent event) {
flwTaskService.setCopy(event.task(), event.flowCopyList());
}
/**
* 发送工作流待办消息。
*
* @param event 工作流待办消息事件
*/
@EventListener
public void handleTaskMessage(WorkflowTaskMessageEvent event) {
flwCommonService.sendMessage(event.flowName(), event.instanceId(), event.messageType(), event.notice());
}
/**
* 发送工作流结果消息。
*
* @param event 工作流结果消息事件
*/
@EventListener
public void handleResultMessage(WorkflowResultMessageEvent event) {
if (!StringUtils.equalsAny(event.status(), BusinessStatusEnum.FINISH.getStatus(), BusinessStatusEnum.BACK.getStatus())) {
return;
}
Long createBy = Convert.toLong(event.createBy(), null);
if (createBy == null) {
return;
}
BusinessStatusEnum status = BusinessStatusEnum.getByStatus(event.status());
if (status == null) {
return;
}
UserDTO initiator = userService.selectById(createBy);
if (initiator == null || initiator.getUserId() == null) {
return;
}
List<String> messageType = CollUtil.isNotEmpty(event.messageType())
? event.messageType()
: Collections.singletonList(MessageTypeEnum.SYSTEM_MESSAGE.getCode());
flwCommonService.sendResultMessage(event.flowName(), status, messageType, Collections.singletonList(initiator));
}
}
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwCommonServiceImpl.java
+29 -23
View File
@@ -11,6 +11,7 @@ import org.dromara.common.core.exception.ServiceException;
import org.dromara.common.core.utils.SpringUtils;
import org.dromara.common.core.utils.StreamUtils;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.core.utils.ThreadUtils;
import org.dromara.common.mail.core.MailBuilder;
import org.dromara.system.api.MessageService;
import org.dromara.system.api.domain.PushPayloadDTO;
@@ -107,23 +108,29 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
Set<String> emails = StreamUtils.toSet(userList, UserDTO::getEmail);
emails.removeIf(StringUtils::isBlank);
for (String code : messageType) {
MessageTypeEnum messageTypeEnum = MessageTypeEnum.getByCode(code);
if (ObjectUtil.isEmpty(messageTypeEnum)) {
continue;
}
try {
switch (messageTypeEnum) {
case SYSTEM_MESSAGE -> {
// 站内消息直接携带前端路由，消息盒子点击后可按路径分流。
messageService.publishMessage(userIds, PushPayloadDTO.of(
PushTypeEnum.MESSAGE,
PushSourceEnum.WORKFLOW,
message, null, path
));
}
case EMAIL_MESSAGE -> MailBuilder.of().to(emails).subject(subject).text(message).send();
case SMS_MESSAGE -> {
Runnable[] sendTasks = messageType.stream()
.map(code -> (Runnable) () -> sendMessageByType(code, message, subject, path, userIds, emails, userList.size()))
.toArray(Runnable[]::new);
ThreadUtils.virtualInvokeAll(sendTasks);
}
private void sendMessageByType(String code, String message, String subject, String path, List<Long> userIds, Set<String> emails, int userCount) {
MessageTypeEnum messageTypeEnum = MessageTypeEnum.getByCode(code);
if (ObjectUtil.isEmpty(messageTypeEnum)) {
return;
}
try {
switch (messageTypeEnum) {
case SYSTEM_MESSAGE -> {
// 站内消息直接携带前端路由，消息盒子点击后可按路径分流。
messageService.publishMessage(userIds, PushPayloadDTO.of(
PushTypeEnum.MESSAGE,
PushSourceEnum.WORKFLOW,
message, null, path
));
}
case EMAIL_MESSAGE -> MailBuilder.of().to(emails).subject(subject).text(message).send();
case SMS_MESSAGE -> {
// LinkedHashMap<String, String> map = new LinkedHashMap<>(1);
// // 根据具体短信服务商参数用法传参
// map.put("code", "1234");
@@ -132,14 +139,13 @@ public class FlwCommonServiceImpl implements IFlwCommonService {
// // 指定获取一个短信服务商 configKey
// SmsBlend smsBlend = SmsFactory.getSmsBlend("config1");
// SmsResponse smsResponse = smsBlend.sendMessage(phones, templateId, map);
log.info("【短信发送 - TODO】用户数量={} 内容={}", userList.size(), message);
}
default -> log.warn("【消息发送】未处理的消息类型：{}", messageTypeEnum);
log.info("【短信发送 - TODO】用户数量={} 内容={}", userCount, message);
}
} catch (Exception ex) {
// 记录错误但不抛出，确保主逻辑不受影响
log.error("【消息发送失败】类型={}，原因={}", messageTypeEnum, ex.getMessage(), ex);
default -> log.warn("【消息发送】未处理的消息类型：{}", messageTypeEnum);
}
} catch (Exception ex) {
// 记录错误但不抛出，确保主逻辑不受影响
log.error("【消息发送失败】类型={}，原因={}", messageTypeEnum, ex.getMessage(), ex);
}
}
ruoyi-modules/ruoyi-workflow/src/main/java/org/dromara/workflow/service/impl/FlwTaskAssigneeServiceImpl.java
+27 -4
View File
@@ -11,6 +11,7 @@ import lombok.extern.slf4j.Slf4j;
import org.dromara.common.core.utils.DateUtils;
import org.dromara.common.core.utils.StreamUtils;
import org.dromara.common.core.utils.StringUtils;
import org.dromara.common.core.utils.ThreadUtils;
import org.dromara.system.api.*;
import org.dromara.system.api.domain.DeptDTO;
import org.dromara.system.api.domain.TaskAssigneeDTO;
@@ -29,6 +30,7 @@ import org.dromara.workflow.service.IFlwTaskAssigneeService;
import org.springframework.stereotype.Service;
import java.util.*;
import java.util.function.Supplier;
import java.util.stream.Collectors;
/**
@@ -103,8 +105,7 @@ public class FlwTaskAssigneeServiceImpl implements IFlwTaskAssigneeService, Hand
}
// 查询所有类型对应的 ID 名称映射
Map<TaskAssigneeEnum, Map<String, String>> nameMap = new EnumMap<>(TaskAssigneeEnum.class);
typeIdMap.forEach((type, ids) -> nameMap.put(type, this.getNamesByType(type, ids)));
Map<TaskAssigneeEnum, Map<String, String>> nameMap = this.getNamesByTypes(typeIdMap);
// 组装返回结果，保持原始顺序
return parsedMap.entrySet().stream()
.map(entry -> {
@@ -213,12 +214,34 @@ public class FlwTaskAssigneeServiceImpl implements IFlwTaskAssigneeService, Hand
typeIdMap.computeIfAbsent(parsed.getKey(), k -> new ArrayList<>()).add(parsed.getValue());
}
}
return typeIdMap.entrySet().stream()
.flatMap(entry -> this.getUsersByType(entry.getKey(), entry.getValue()).stream())
return this.getUsersByTypes(typeIdMap).stream()
.distinct()
.toList();
}
private List<UserDTO> getUsersByTypes(Map<TaskAssigneeEnum, List<String>> typeIdMap) {
List<Supplier<List<UserDTO>>> suppliers = typeIdMap.entrySet().stream()
.map(entry -> (Supplier<List<UserDTO>>) () -> this.getUsersByType(entry.getKey(), entry.getValue()))
.toList();
return ThreadUtils.virtualSubmitAll(suppliers).stream()
.filter(CollUtil::isNotEmpty)
.flatMap(Collection::stream)
.toList();
}
private Map<TaskAssigneeEnum, Map<String, String>> getNamesByTypes(Map<TaskAssigneeEnum, List<String>> typeIdMap) {
List<TaskAssigneeEnum> types = new ArrayList<>(typeIdMap.keySet());
List<Supplier<Map<String, String>>> suppliers = types.stream()
.map(type -> (Supplier<Map<String, String>>) () -> this.getNamesByType(type, typeIdMap.get(type)))
.toList();
List<Map<String, String>> names = ThreadUtils.virtualSubmitAll(suppliers);
Map<TaskAssigneeEnum, Map<String, String>> nameMap = new EnumMap<>(TaskAssigneeEnum.class);
for (int i = 0; i < types.size(); i++) {
nameMap.put(types.get(i), names.get(i));
}
return nameMap;
}
/**
* 根据指定的任务分配类型（TaskAssigneeEnum）和 ID 列表，获取对应的用户信息列表
*