项目初始化

datax-common/datax-common-rabbitmq/src/main/java/cn/datax/common/rabbitmq/config/RabbitMqConstant.java

@@ -25,5 +25,5 @@ public class RabbitMqConstant {
     /**
      * TOPIC类型的路由键：工作流 {}占位符替换
      */
-    public static final String TOPIC_WORKFLOW_KEY = "topic.workflow.key.{}";
+    public static final String TOPIC_WORKFLOW_KEY = "topic.workflow.key.";
 }

datax-config/src/main/resources/config/datax-service-data-masterdata-dev.yml

@@ -12,6 +12,16 @@ spring:
         max-wait: -1ms    # 连接池最大阻塞等待时间（使用负值表示没有限制）
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 5       # 连接池中的最小空闲连接
+  rabbitmq:
+    host: localhost
+    port: 5672
+    username: admin
+    password: 1234@abcd
+    listener:
+      simple:
+        acknowledge-mode: manual
+        concurrency: 1
+        max-concurrency: 10
   datasource:
     dynamic:
       primary: mysql

datax-config/src/main/resources/config/datax-service-workflow-dev.yml

@@ -12,6 +12,15 @@ spring:
         max-wait: -1ms    # 连接池最大阻塞等待时间（使用负值表示没有限制）
         max-idle: 10      # 连接池中的最大空闲连接
         min-idle: 5       # 连接池中的最小空闲连接
+  rabbitmq:
+    host: localhost
+    port: 5672
+    username: admin
+    password: 1234@abcd
+    publisher-confirm-type: correlated
+    publisher-returns: true
+    template:
+      mandatory: true
   datasource:
     dynamic:
       primary: mysql

datax-modules/data-market-service-parent/data-market-service-mapping/src/main/java/cn/datax/service/data/market/mapping/config/RabbitMqListenerConfig.java

@@ -2,7 +2,9 @@ package cn.datax.service.data.market.mapping.config;
 
 import cn.datax.common.rabbitmq.config.RabbitMqConstant;
 import cn.datax.common.utils.ThrowableUtil;
-import cn.datax.service.data.market.mapping.service.QueueHandlerService;
+import cn.datax.service.data.market.api.entity.DataApiEntity;
+import cn.datax.service.data.market.api.feign.DataApiServiceFeign;
+import cn.datax.service.data.market.mapping.handler.MappingHandlerMapping;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.Message;
@@ -19,12 +21,15 @@ import java.util.Map;
 @Configuration
 public class RabbitMqListenerConfig {
 
-    @Autowired
-    private QueueHandlerService queueHandlerService;
-
     private static String HANDLER_RELEASE = "1";
     private static String HANDLER_CANCEL = "2";
 
+    @Autowired
+    private DataApiServiceFeign dataApiServiceFeign;
+
+    @Autowired
+    private MappingHandlerMapping mappingHandlerMapping;
+
     /**
      * api发布与撤销
      * @param map type 1:发布 2:撤销
@@ -40,10 +45,13 @@ public class RabbitMqListenerConfig {
             String id = (String) map.get("id");
             String type = (String) map.get("type");
             log.info("fanoutQueueRelease接收到了：{},{}", id, type);
-            if (HANDLER_RELEASE.equals(type)) {
-                queueHandlerService.handlerRelease(id);
-            } else if (HANDLER_CANCEL.equals(type)) {
-                queueHandlerService.handlerCancel(id);
+            DataApiEntity dataApiEntity = dataApiServiceFeign.getDataApiById(id);
+            if (dataApiEntity != null) {
+                if (HANDLER_RELEASE.equals(type)) {
+                    mappingHandlerMapping.registerMapping(dataApiEntity);
+                } else if (HANDLER_CANCEL.equals(type)) {
+                    mappingHandlerMapping.unregisterMapping(dataApiEntity);
+                }
             }
             // 手动确认
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
@@ -56,7 +64,7 @@ public class RabbitMqListenerConfig {
                 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
             }else {
                 //记录日志
-                log.error("消息消费失败处理：{}",e.getMessage());
+                log.error("消息消费失败处理：{}", e.getMessage());
                 //第一个参数为消息的index，第二个参数是是否批量处理，第三个参数为是否让被拒绝的消息重新入队列
                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
             }

datax-modules/data-market-service-parent/data-market-service-mapping/src/main/java/cn/datax/service/data/market/mapping/service/QueueHandlerService.java

-package cn.datax.service.data.market.mapping.service;
-
-public interface QueueHandlerService {
-
-    void handlerRelease(String id);
-
-    void handlerCancel(String id);
-}

datax-modules/data-market-service-parent/data-market-service-mapping/src/main/java/cn/datax/service/data/market/mapping/service/impl/QueueHandlerServiceImpl.java

-package cn.datax.service.data.market.mapping.service.impl;
-
-import cn.datax.service.data.market.api.entity.DataApiEntity;
-import cn.datax.service.data.market.api.feign.DataApiServiceFeign;
-import cn.datax.service.data.market.mapping.handler.MappingHandlerMapping;
-import cn.datax.service.data.market.mapping.service.QueueHandlerService;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-@Slf4j
-@Service
-public class QueueHandlerServiceImpl implements QueueHandlerService {
-
-    @Autowired
-    private DataApiServiceFeign dataApiServiceFeign;
-
-    @Autowired
-    private MappingHandlerMapping mappingHandlerMapping;
-
-    @Override
-    public void handlerRelease(String id) {
-        DataApiEntity dataApiEntity = dataApiServiceFeign.getDataApiById(id);
-        if (dataApiEntity != null) {
-            mappingHandlerMapping.registerMapping(dataApiEntity);
-        }
-    }
-
-    @Override
-    public void handlerCancel(String id) {
-        DataApiEntity dataApiEntity = dataApiServiceFeign.getDataApiById(id);
-        if (dataApiEntity != null) {
-            mappingHandlerMapping.unregisterMapping(dataApiEntity);
-        }
-    }
-}

datax-modules/data-market-service-parent/data-market-service/src/main/java/cn/datax/service/data/market/service/impl/DataApiServiceImpl.java

@@ -363,7 +363,7 @@ public class DataApiServiceImpl extends BaseServiceImpl<DataApiDao, DataApiEntit
 
     @Override
     public void releaseDataApi(String id) {
-        Map<String, Object> map = new HashMap<>();
+        Map<String, Object> map = new HashMap<>(2);
         map.put("id", id);
         map.put("type", "1");
         String obj = (String) Optional.ofNullable(rabbitTemplate.convertSendAndReceive(RabbitMqConstant.FANOUT_API_QUEUE, "", map)).orElse("");
@@ -377,7 +377,7 @@ public class DataApiServiceImpl extends BaseServiceImpl<DataApiDao, DataApiEntit
 
     @Override
     public void cancelDataApi(String id) {
-        Map<String, Object> map = new HashMap<>();
+        Map<String, Object> map = new HashMap<>(2);
         map.put("id", id);
         map.put("type", "2");
         String obj = (String) Optional.ofNullable(rabbitTemplate.convertSendAndReceive(RabbitMqConstant.FANOUT_API_QUEUE, "", map)).orElse("");

datax-modules/data-masterdata-service-parent/data-masterdata-service/src/main/java/cn/datax/service/data/masterdata/config/RabbitMqListenerConfig.java

@@ -2,6 +2,9 @@ package cn.datax.service.data.masterdata.config;
 
 import cn.datax.common.rabbitmq.config.RabbitMqConstant;
 import cn.datax.common.utils.ThrowableUtil;
+import cn.datax.service.data.masterdata.api.entity.ModelEntity;
+import cn.datax.service.data.masterdata.dao.ModelDao;
+import cn.datax.service.workflow.api.enums.VariablesEnum;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.amqp.core.Message;
@@ -9,29 +12,41 @@ import org.springframework.amqp.rabbit.annotation.Exchange;
 import org.springframework.amqp.rabbit.annotation.Queue;
 import org.springframework.amqp.rabbit.annotation.QueueBinding;
 import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.Map;
+
 @Slf4j
 @Configuration
 public class RabbitMqListenerConfig {
 
+    @Autowired
+    private ModelDao modelDao;
+
     /**
      * 消费工作流 业务编码 6011
-     * @param id
+     * @param map
      * @param channel
      * @param message
      * @return
      * @throws Exception
      */
     @RabbitListener(bindings = @QueueBinding(exchange = @Exchange(name = RabbitMqConstant.TOPIC_EXCHANGE_WORKFLOW, type = "topic", durable = "true", autoDelete = "false"),
-            key = {""},
+            key = { RabbitMqConstant.TOPIC_WORKFLOW_KEY + "6011" },
             value = @Queue(value = RabbitMqConstant.TOPIC_WORKFLOW_QUEUE, durable = "true", exclusive = "false", autoDelete = "false")))
-    public String fanoutQueueRelease(String id, Channel channel, Message message) throws Exception {
+    public void fanoutQueueRelease(Map map, Channel channel, Message message) throws Exception {
         try {
-            log.info("fanoutQueueRelease接收到了：{}", id);
+            log.info("接收到了消息：{}", map);
+            String businessKey = (String) map.get(VariablesEnum.businessKey.toString());
+            String businessCode = (String) map.get(VariablesEnum.businessCode.toString());
+            String flowStatus = (String) map.get("flowStatus");
+            ModelEntity model = new ModelEntity();
+            model.setId(businessKey);
+            model.setFlowStatus(flowStatus);
+            modelDao.updateById(model);
             // 手动确认
             channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
-            return id;
         }catch (Exception e){
             log.error("全局异常信息ex={}, StackTrace={}", e.getMessage(), ThrowableUtil.getStackTrace(e));
             if (message.getMessageProperties().getRedelivered()){
@@ -40,11 +55,10 @@ public class RabbitMqListenerConfig {
                 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
             }else {
                 //记录日志
-                log.error("消息消费失败处理：{}",e.getMessage());
+                log.error("消息消费失败处理：{}", e.getMessage());
                 //第一个参数为消息的index，第二个参数是是否批量处理，第三个参数为是否让被拒绝的消息重新入队列
                 channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
             }
         }
-        return null;
     }
 }

datax-modules/workflow-service-parent/workflow-service/src/main/java/cn/datax/service/workflow/config/RabbitConfig.java

+package cn.datax.service.workflow.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Slf4j
+@Configuration
+public class RabbitConfig {
+
+    @Bean
+    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
+        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
+        // 消息是否成功发送到Exchange
+        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
+            if (ack) {
+                log.info("消息成功发送到Exchange");
+            } else {
+                log.info("消息发送到Exchange失败, {}, cause: {}", correlationData, cause);
+            }
+        });
+        // 触发setReturnCallback回调必须设置mandatory=true, 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
+        rabbitTemplate.setMandatory(true);
+        // 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
+        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
+            log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}", exchange, routingKey, replyCode, replyText, message);
+        });
+        return rabbitTemplate;
+    }
+}

datax-modules/workflow-service-parent/workflow-service/src/main/java/cn/datax/service/workflow/flowable/EndTaskListener.java

 package cn.datax.service.workflow.flowable;
 
 import cn.datax.common.core.DataConstant;
+import cn.datax.common.rabbitmq.config.RabbitMqConstant;
+import cn.datax.common.utils.SpringContextHolder;
 import cn.datax.service.workflow.api.enums.VariablesEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.engine.delegate.DelegateExecution;
 import org.flowable.engine.delegate.ExecutionListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
+import java.util.HashMap;
 import java.util.Map;
 
 @Slf4j
@@ -22,10 +26,17 @@ public class EndTaskListener implements ExecutionListener {
         String businessKey = (String) variables.get(VariablesEnum.businessKey.toString());
         String businessCode = (String) variables.get(VariablesEnum.businessCode.toString());
         log.info("业务结束：{},{}", businessKey, businessCode);
+        Map<String, Object> map = new HashMap<>(4);
+        map.put(VariablesEnum.businessKey.toString(), businessKey);
+        map.put(VariablesEnum.businessCode.toString(), businessCode);
         if (delegateExecution.getCurrentActivityId().equals(VariablesEnum.approveEnd.toString())) {
             log.info("业务结束状态：{}", DataConstant.AuditState.AGREE.getKey());
+            map.put("flowStatus", DataConstant.AuditState.AGREE.getKey());
         } else if (delegateExecution.getCurrentActivityId().equals(VariablesEnum.rejectEnd.toString())) {
             log.info("业务结束状态：{}", DataConstant.AuditState.REJECT.getKey());
+            map.put("flowStatus", DataConstant.AuditState.REJECT.getKey());
         }
+        RabbitTemplate rabbitTemplate = SpringContextHolder.getBean(RabbitTemplate.class);
+        rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE_WORKFLOW, RabbitMqConstant.TOPIC_WORKFLOW_KEY + businessCode, map);
     }
 }

datax-modules/workflow-service-parent/workflow-service/src/main/java/cn/datax/service/workflow/flowable/InitialAuditCompleteTaskListener.java

 package cn.datax.service.workflow.flowable;
 
+import cn.datax.common.core.DataConstant;
+import cn.datax.common.rabbitmq.config.RabbitMqConstant;
 import cn.datax.common.utils.SpringContextHolder;
 import cn.datax.service.workflow.api.enums.VariablesEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.engine.TaskService;
 import org.flowable.task.service.delegate.DelegateTask;
 import org.flowable.task.service.delegate.TaskListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
+import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
 
@@ -28,6 +32,13 @@ public class InitialAuditCompleteTaskListener implements TaskListener {
             String businessKey = (String) variables.get(VariablesEnum.businessKey.toString());
             String businessCode = (String) variables.get(VariablesEnum.businessCode.toString());
             log.info("业务退回：{},{}", businessKey, businessCode);
+            log.info("业务退回状态：{}", DataConstant.AuditState.BACK.getKey());
+            RabbitTemplate rabbitTemplate = SpringContextHolder.getBean(RabbitTemplate.class);
+            Map<String, Object> map = new HashMap<>(4);
+            map.put(VariablesEnum.businessKey.toString(), businessKey);
+            map.put(VariablesEnum.businessCode.toString(), businessCode);
+            map.put("flowStatus", DataConstant.AuditState.BACK.getKey());
+            rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE_WORKFLOW, RabbitMqConstant.TOPIC_WORKFLOW_KEY + businessCode, map);
         }
         log.info("退出初审节点用户任务完成监听器");
     }

datax-modules/workflow-service-parent/workflow-service/src/main/java/cn/datax/service/workflow/flowable/SubmitCompleteTaskListener.java

 package cn.datax.service.workflow.flowable;
 
 import cn.datax.common.core.DataConstant;
+import cn.datax.common.rabbitmq.config.RabbitMqConstant;
+import cn.datax.common.utils.SpringContextHolder;
 import cn.datax.service.workflow.api.enums.VariablesEnum;
 import lombok.extern.slf4j.Slf4j;
 import org.flowable.task.service.delegate.DelegateTask;
 import org.flowable.task.service.delegate.TaskListener;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 
+import java.util.HashMap;
 import java.util.Map;
 
 @Slf4j
@@ -25,5 +29,11 @@ public class SubmitCompleteTaskListener implements TaskListener {
         log.info("业务审核中：{},{}", businessKey, businessCode);
         log.info("业务审核中状态：{}", DataConstant.AuditState.AUDIT.getKey());
         log.info("退出提交节点用户任务完成监听器");
+        RabbitTemplate rabbitTemplate = SpringContextHolder.getBean(RabbitTemplate.class);
+        Map<String, Object> map = new HashMap<>(4);
+        map.put(VariablesEnum.businessKey.toString(), businessKey);
+        map.put(VariablesEnum.businessCode.toString(), businessCode);
+        map.put("flowStatus", DataConstant.AuditState.AUDIT.getKey());
+        rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE_WORKFLOW, RabbitMqConstant.TOPIC_WORKFLOW_KEY + businessCode, map);
     }
 }

datax-modules/workflow-service-parent/workflow-service/src/main/java/cn/datax/service/workflow/service/impl/FlowInstanceServiceImpl.java

 package cn.datax.service.workflow.service.impl;
 
+import cn.datax.common.core.DataConstant;
+import cn.datax.common.rabbitmq.config.RabbitMqConstant;
 import cn.datax.common.utils.SecurityUtil;
 import cn.datax.service.workflow.api.dto.ProcessInstanceCreateRequest;
 import cn.datax.service.workflow.api.enums.VariablesEnum;
@@ -27,12 +29,14 @@ import org.flowable.engine.runtime.ProcessInstanceQuery;
 import org.flowable.engine.task.Comment;
 import org.flowable.image.impl.DefaultProcessDiagramGenerator;
 import org.flowable.task.api.Task;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
 
 import java.io.InputStream;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -52,6 +56,9 @@ public class FlowInstanceServiceImpl implements FlowInstanceService {
     @Autowired
     private TaskService taskService;
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
     private static final String IMAGE_TYPE = "png";
     private static final String FONT_NAME = "宋体";
 
@@ -103,13 +110,19 @@ public class FlowInstanceServiceImpl implements FlowInstanceService {
     @Override
     public void deleteProcessInstance(String processInstanceId) {
         // 发送消息队列
-        ProcessInstance processInstance = runtimeService.createProcessInstanceQuery().processInstanceId(processInstanceId).singleResult();
+        ProcessInstance processInstance = runtimeService.createProcessInstanceQuery().includeProcessVariables().processInstanceId(processInstanceId).singleResult();
         Map<String, Object> variables = processInstance.getProcessVariables();
         String businessKey = (String) variables.get(VariablesEnum.businessKey.toString());
         String businessCode = (String) variables.get(VariablesEnum.businessCode.toString());
         log.info("业务撤销：{},{}", businessKey, businessCode);
         log.info("成功删除流程实例ID:{}", processInstanceId);
+        log.info("业务撤销状态：{}", DataConstant.AuditState.CANCEL.getKey());
         runtimeService.deleteProcessInstance(processInstanceId, "用户撤销");
+        Map<String, Object> map = new HashMap<>(4);
+        map.put(VariablesEnum.businessKey.toString(), businessKey);
+        map.put(VariablesEnum.businessCode.toString(), businessCode);
+        map.put("flowStatus", DataConstant.AuditState.CANCEL.getKey());
+        rabbitTemplate.convertAndSend(RabbitMqConstant.TOPIC_EXCHANGE_WORKFLOW, RabbitMqConstant.TOPIC_WORKFLOW_KEY + businessCode, map);
     }
 
     @Override