feat(pigx-flow-api): 新增流程状态事件发布订阅机制

- 新增 FlowStatusEventPublisher 用于发布流程状态变更事件到 Redis Channel
- 新增 FlowStatusEventListener 用于监听 Redis 消息并分发给匹配的事件服务
- 新增 IProcessInstanceStatusEventService 接口定义流程状态事件处理规范
- 新增 FlowStatusEventAutoConfiguration 自动配置类,实现事件发布者和监听器的 Bean 注册
- 重构 NodeTaskCompleteNotify 服务,使用事件发布机制替代直接调用事件服务
- 将 IProcessInstanceStatusEventService 接口移动到 flow-api 模块的 event 包下
- 在 pigx-flow-api 模块添加 spring-boot-starter-data-redis 依赖支持
- 更新相关服务的包引用和日期注释
This commit is contained in:
PIGCLOUD committed 2026-03-08 15:02:40 +08:00
1 parent d1858d5844
commit 2754607924
10 files changed
+214 -31

No files matched your search

@@ -0,0 +1,58 @@
package com.pig4cloud.pigx.flow.config;
import com.pig4cloud.pigx.flow.event.FlowStatusEventListener;
import com.pig4cloud.pigx.flow.event.FlowStatusEventPublisher;
import com.pig4cloud.pigx.flow.event.IProcessInstanceStatusEventService;
import org.springframework.boot.autoconfigure.AutoConfiguration;
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.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import java.util.List;
/**
* 流程状态事件自动配置
*
* @author lengleng
* @date 2026-02-10
*/
@AutoConfiguration
@ConditionalOnClass(StringRedisTemplate.class)
public class FlowStatusEventAutoConfiguration {
/**
* 事件发布者 - 所有依赖 flow-api 的模块可用
*/
@Bean
@ConditionalOnMissingBean
public FlowStatusEventPublisher flowStatusEventPublisher(StringRedisTemplate redisTemplate) {
return new FlowStatusEventPublisher(redisTemplate);
}
/**
* 事件监听器 - 仅当有 IProcessInstanceStatusEventService 实现时启用
*/
@Bean
@ConditionalOnMissingBean
@ConditionalOnBean(IProcessInstanceStatusEventService.class)
public FlowStatusEventListener flowStatusEventListener(List<IProcessInstanceStatusEventService> eventServices,
StringRedisTemplate redisTemplate) {
return new FlowStatusEventListener(eventServices, redisTemplate);
}
/**
* 注册监听器到 Redis 容器
*/
@Bean
@ConditionalOnBean(FlowStatusEventListener.class)
public ChannelTopic flowStatusTopic(RedisMessageListenerContainer container, FlowStatusEventListener listener) {
ChannelTopic topic = new ChannelTopic(FlowStatusEventPublisher.CHANNEL);
container.addMessageListener(listener, topic);
return topic;
}
}
@@ -0,0 +1,52 @@
package com.pig4cloud.pigx.flow.event;
import cn.hutool.json.JSONUtil;
import com.pig4cloud.pigx.flow.dto.ProcessInstanceParamDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.util.List;
/**
* 流程状态事件监听器
* <p>
* 监听 Redis Channel,将事件分发给匹配的 IProcessInstanceStatusEventService 实现
* </p>
*
* @author lengleng
* @date 2026-02-10
*/
@Slf4j
@RequiredArgsConstructor
public class FlowStatusEventListener implements MessageListener {
private final List<IProcessInstanceStatusEventService> eventServices;
private final StringRedisTemplate redisTemplate;
@Override
public void onMessage(Message message, byte[] pattern) {
String body = redisTemplate.getStringSerializer().deserialize(message.getBody());
ProcessInstanceParamDto paramDto = JSONUtil.toBean(body, ProcessInstanceParamDto.class);
log.debug("收到流程状态事件: flowId={}, processInstanceId={}", paramDto.getFlowId(),
paramDto.getProcessInstanceId());
// 分发给匹配 flowId 的实现
for (IProcessInstanceStatusEventService service : eventServices) {
if (service.getFlowId().equals(paramDto.getFlowId())) {
try {
service.handleStatusEvent(paramDto);
}
catch (Exception e) {
log.error("处理流程状态事件失败: flowId={}, service={}", paramDto.getFlowId(),
service.getClass().getSimpleName(), e);
}
}
}
}
}
@@ -0,0 +1,37 @@
package com.pig4cloud.pigx.flow.event;
import cn.hutool.json.JSONUtil;
import com.pig4cloud.pigx.flow.dto.ProcessInstanceParamDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
/**
* 流程状态事件发布者
* <p>
* 将流程状态变更事件发布到 Redis Channel
* </p>
*
* @author lengleng
* @date 2026-02-10
*/
@Slf4j
@RequiredArgsConstructor
public class FlowStatusEventPublisher {
public static final String CHANNEL = "pigx:flow:status";
private final StringRedisTemplate redisTemplate;
/**
* 发布流程状态变更事件
* @param paramDto 流程实例参数
*/
public void publish(ProcessInstanceParamDto paramDto) {
String message = JSONUtil.toJsonStr(paramDto);
redisTemplate.convertAndSend(CHANNEL, message);
log.debug("发布流程状态事件: flowId={}, processInstanceId={}", paramDto.getFlowId(),
paramDto.getProcessInstanceId());
}
}
@@ -0,0 +1,29 @@
package com.pig4cloud.pigx.flow.event;
import com.pig4cloud.pigx.flow.dto.ProcessInstanceParamDto;
/**
* 流程实例状态事件服务接口
* <p>
* 实现此接口以监听流程状态变更事件。
* 通过 Redis Pub/Sub 实现跨模块通知。
* </p>
*
* @author lengleng
* @date 2026-02-10
*/
public interface IProcessInstanceStatusEventService {
/**
* 获取监听的流程定义Key
* @return 监听的流程定义Key
*/
String getFlowId();
/**
* 处理状态事件
* @param paramDto 流程实例参数DTO
*/
void handleStatusEvent(ProcessInstanceParamDto paramDto);
}