update - 优化特定场景下扩展任务表的状态不太正常的问题

workflow-engine-core/src/main/java/cn/axzo/workflow/core/repository/mapper/CommonMapper.java

@@ -0,0 +1,18 @@
+package cn.axzo.workflow.core.repository.mapper;
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * TODO
+ *
+ * @author wangli
+ * @since 2024/4/28 14:44
+ */
+@Mapper
+public interface CommonMapper {
+    List<Map<String, Object>> executeDynamicSQL(@Param("sql") String sql);
+}

workflow-engine-core/src/main/java/cn/axzo/workflow/core/service/impl/ExtAxHiTaskInstServiceImpl.java

@@ -15,6 +15,8 @@ import javax.annotation.Resource;
 import java.util.List;
 import java.util.Objects;
 
+import static cn.axzo.workflow.common.enums.BpmnProcessInstanceResultEnum.PROCESSING;
+
 /**
  * 流程定义扩展表操作服务实现
  *
@@ -66,6 +68,14 @@ public class ExtAxHiTaskInstServiceImpl implements ExtAxHiTaskInstService {
 
     @Override
     public void updateByTaskIdAndInstanceId(String taskId, String processInstanceId, String assignee, BpmnProcessInstanceResultEnum resultEnum) {
+        ExtAxHiTaskInst existence = getByTaskId(taskId, processInstanceId);
+        if (Objects.nonNull(existence)) {
+            log.info("更新扩展任务实例表数据: taskId:{}, instanceId:{}, currentStatus:{}, anticipateStatus: {}",
+                    taskId, processInstanceId, existence.getStatus(), resultEnum.getStatus());
+        } else {
+            log.warn("无法正确更新扩展任务表数据: taskId:{}, instanceId:{}, anticipateStatus: {}",
+                    taskId, processInstanceId, resultEnum.getStatus());
+        }
         ExtAxHiTaskInst entity = new ExtAxHiTaskInst();
         if (StringUtils.hasText(assignee)) {
             entity.setAssignee(assignee);
@@ -74,7 +84,9 @@ public class ExtAxHiTaskInstServiceImpl implements ExtAxHiTaskInstService {
 
         LambdaUpdateWrapper<ExtAxHiTaskInst> wrapper = new LambdaUpdateWrapper<ExtAxHiTaskInst>()
                 .eq(ExtAxHiTaskInst::getTaskId, taskId)
-                .eq(ExtAxHiTaskInst::getProcInstId, processInstanceId);
+                .eq(ExtAxHiTaskInst::getProcInstId, processInstanceId)
+                // 可能由于事件消费时, 嵌套了引擎操作, 会引起一部分的消息顺序异常, 导致状态被异常变更
+                .eq(ExtAxHiTaskInst::getStatus, PROCESSING.getStatus());
         extAxHiTaskInstMapper.update(entity, wrapper);
     }
 }

workflow-engine-core/src/main/resources/mapper/common.xml

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+
+<mapper namespace="cn.axzo.workflow.core.repository.mapper.CommonMapper">
+    <!-- 定义执行动态 SQL 的方法 -->
+    <select id="executeDynamicSQL" resultType="java.util.Map">
+        ${sql}
+    </select>
+</mapper>

workflow-engine-server/pom.xml

@@ -115,6 +115,10 @@
             <groupId>org.apache.maven</groupId>
             <artifactId>maven-artifact</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.xuxueli</groupId>
+            <artifactId>xxl-job-core</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

workflow-engine-server/src/main/java/cn/axzo/workflow/server/common/config/WebMvcConfiguration.java

@@ -13,7 +13,7 @@ import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
  * @since 2024/2/6 11:35
  */
 @Configuration(proxyBeanMethods = false)
-public class WebMvcConfig implements WebMvcConfigurer {
+public class WebMvcConfiguration implements WebMvcConfigurer {
     @Autowired
     private RequestHeaderContextInterceptor requestHeaderContextInterceptor;
 

workflow-engine-server/src/main/java/cn/axzo/workflow/server/common/config/XxlJobConfiguration.java

@@ -0,0 +1,56 @@
+package cn.axzo.workflow.server.common.config;
+
+import cn.azxo.framework.common.annotation.OnlyPodsEnvironment;
+import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * TODO
+ *
+ * @author wangli
+ * @since 2024/4/28 14:02
+ */
+@Configuration
+public class XxlJobConfiguration {
+    Logger logger = LoggerFactory.getLogger(XxlJobConfiguration.class);
+
+    @Value("${xxl.job.admin.addresses}")
+    private String adminAddresses;
+
+    @Value("${xxl.job.executor.appname}")
+    private String appName;
+
+    @Value("")
+    private String ip;
+
+    @Value("${xxl.job.executor.port}")
+    private int port;
+
+    @Value("")
+    private String accessToken;
+
+    @Value("")
+    private String logPath;
+
+    @Value("-1")
+    private int logRetentionDays;
+
+    @Bean
+    @OnlyPodsEnvironment
+    public XxlJobSpringExecutor xxlJobExecutor() {
+        logger.info(">>>>>>>>>>> xxl-job config init.");
+        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
+        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
+        xxlJobSpringExecutor.setAppname(appName);
+        xxlJobSpringExecutor.setIp(ip);
+        xxlJobSpringExecutor.setPort(port);
+        xxlJobSpringExecutor.setAccessToken(accessToken);
+        xxlJobSpringExecutor.setLogPath(logPath);
+        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
+        return xxlJobSpringExecutor;
+    }
+}

workflow-engine-server/src/main/java/cn/axzo/workflow/server/xxljob/OperationDataJobHandler.java

@@ -0,0 +1,37 @@
+package cn.axzo.workflow.server.xxljob;
+
+import cn.axzo.framework.jackson.utility.JSON;
+import cn.axzo.workflow.core.repository.mapper.CommonMapper;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.IJobHandler;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import com.xxl.job.core.log.XxlJobLogger;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 操作数据表
+ *
+ * @author wangli
+ * @since 2024/4/28 14:06
+ */
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class OperationDataJobHandler extends IJobHandler {
+    @Resource
+    private CommonMapper commonMapper;
+
+    @Override
+    @XxlJob("executeDynamicSql")
+    public ReturnT<String> execute(String s) throws Exception {
+        List<Map<String, Object>> maps = commonMapper.executeDynamicSQL(s);
+        XxlJobLogger.log("result: {}", JSON.toJSONString(maps));
+        return ReturnT.SUCCESS;
+    }
+}