update:增加使用mq+数据库时的构建的场景

ibzdst-core/src/main/java/cn/ibizlab/core/extensions/service/AnalyseSqlServiceImpl.java

@@ -17,9 +17,11 @@ import cn.ibizlab.util.dict.CodeItem;
 import cn.ibizlab.util.dict.CodeList;
 import cn.ibizlab.util.helper.CachedBeanCopier;
 import cn.ibizlab.util.helper.DataObject;
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
 import lombok.extern.slf4j.Slf4j;
 import org.kie.api.runtime.KieSession;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
@@ -42,6 +44,9 @@ public class AnalyseSqlServiceImpl extends BaseEntityServiceImpl {
     @Lazy
     private DABuildExService daBuildExService;
 
+    @Value("${rocketmq.sendResultsBatchSize:1000}")
+    private Integer sendResultsBatchSize;
+
     @Override
     public void saveResult(ModelObj param, String RULEID,String RULECODE, String RULENAME, String RU_EXECRESULTNAME, FieldObj BUSINESSCAT,
                            Integer RETVALUE, FieldObj KEYVALUEFIELD, FieldObj DOMAINSFIELD,FieldObj DIMFIELD,
@@ -254,6 +259,14 @@ public class AnalyseSqlServiceImpl extends BaseEntityServiceImpl {
                     if(deleteResults != null && deleteResults instanceof List){
                         deleteDatas.addAll((List<ExecResult>)deleteResults);
                     }
+                    if (saveDatas.size() > sendResultsBatchSize || deleteDatas.size() > sendResultsBatchSize) {
+                        resultsMQMsg.setSaveDatas(saveDatas);
+                        resultsMQMsg.setDeleteDatas(deleteDatas);
+                        defaultMQProducerService.sendSqlResultsMsg(resultsMQMsg);
+                        resultsMQMsg.setMsgId(IdWorker.getIdStr());
+                        saveDatas.clear();
+                        deleteDatas.clear();
+                    }
                 }
                 catch (Exception e)
                 {
@@ -273,13 +286,11 @@ public class AnalyseSqlServiceImpl extends BaseEntityServiceImpl {
             }
 
             //发送构建结果消息
-            if(saveDatas.size() > 0) {
+            if(saveDatas.size() > 0 || deleteDatas.size() > 0) {
                 resultsMQMsg.setSaveDatas(saveDatas);
-            }
-            if(deleteDatas.size() > 0)
                 resultsMQMsg.setDeleteDatas(deleteDatas);
-            defaultMQProducerService.sendSqlResultsMsg(resultsMQMsg);
-
+                defaultMQProducerService.sendSqlResultsMsg(resultsMQMsg);
+            }
         }
         catch(Exception e)
         {

ibzdst-core/src/main/java/cn/ibizlab/core/extensions/util/DefaultMQProducerService.java

@@ -103,6 +103,20 @@ public class DefaultMQProducerService implements MsgProducerService{
 
     @Override
     public void sendSqlResultsMsg(SqlResultsMQMsg resultsMQMsg) throws Exception {
+        String msg = JSON.toJSONString(resultsMQMsg);
+        log.trace(String.format("sendResultsMsg:%1$s", msg));
+        dstRocketMQTemplate.asyncSendOrderly(resultsTopic+":SqlResult",msg, DigestUtils.md5DigestAsHex(msg.getBytes()),new SendCallback(){
+            @Override
+            public void onSuccess(SendResult sendResult) {
+                log.info("结果消息发送响应：" + sendResult.toString());
+            }
+
+            @Override
+            public void onException(Throwable throwable) {
+                throwable.printStackTrace();
+            }
+
+        });
 
     }
 

ibzdst-core/src/main/java/cn/ibizlab/core/message/ResultMQMsgListener.java

@@ -2,6 +2,12 @@ package cn.ibizlab.core.message;
 
 import cn.ibizlab.core.extensions.cql.ExecResultRepository;
 import cn.ibizlab.core.extensions.domain.ResultsMQMsg;
+import cn.ibizlab.core.extensions.domain.SqlResultsMQMsg;
+import cn.ibizlab.core.lite.extensions.service.LiteDataService;
+import cn.ibizlab.core.rule.domain.ExecLog;
+import cn.ibizlab.core.rule.domain.ExecResult;
+import cn.ibizlab.core.rule.service.IExecLogService;
+import cn.ibizlab.util.helper.DataObject;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.TypeReference;
 import lombok.extern.slf4j.Slf4j;
@@ -10,11 +16,14 @@ import org.apache.rocketmq.spring.annotation.ConsumeMode;
 import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
 import org.apache.rocketmq.spring.core.RocketMQListener;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Component;
 import org.springframework.util.StringUtils;
 
+import java.util.List;
+
 @Slf4j
 @Component
 @ConditionalOnExpression("'${rocketmq.consumer.namesrvAddr:}'!='' && ${rocketmq.consumer.enabled:false}")
@@ -31,18 +40,75 @@ public class ResultMQMsgListener implements RocketMQListener<MessageExt> {
     @Lazy
     private ExecResultRepository execResultRepository;
 
+    @Value("${rocketmq.producer.resultsTopic:DSTRESULTSMSG}")
+    private String resultsTopic;
+
+    @Value("${defaultResultDataSource:default}")
+    private String Default_ResultDataSource;
+
+    @Value("${defaultResultTableName:IBZRULERESULT}")
+    private String Default_ResultTableName;
+
+    @Value("${ibiz.saveResultBatchSize:2000}")
+    private int saveResultBatchSize;
+
+    @Value("${ibiz.deleteResultBatchBatchSize:20000}")
+    private int deleteResultBatchSize;
+
+    @Autowired(required = false)
+    @Lazy
+    private cn.ibizlab.core.extensions.service.ExecResultExService execResultExService;
+
+    @Value("${cassandra.host:}")
+    private String cassandraHost;
+
+    @Autowired
+    @Lazy
+    protected IExecLogService ruExecLogService;
 
     @Override
     public void onMessage(MessageExt messageExt) {
 
         try {
-            log.info("MQ消息topic={}, tags={}, 消息内容={}", messageExt.getTopic(), messageExt.getTags(), messageExt.getMsgId());
             String body = new String(messageExt.getBody(), "utf-8");
             String tags = messageExt.getTags();
-
-            ResultsMQMsg resultsMQMsg = JSON.parseObject(body, new TypeReference<ResultsMQMsg>() {});
-            if(!StringUtils.isEmpty(resultsMQMsg.getKeyValueField()))
-                execResultRepository.saveResultsMQMsg(resultsMQMsg);
+            String topic = messageExt.getTopic();
+            log.info("MQ消息topic={}, tags={}, 消息内容={}", topic, tags, messageExt.getMsgId());
+            if (resultsTopic.equalsIgnoreCase(topic)) {
+                if (StringUtils.isEmpty(cassandraHost)) {
+                    SqlResultsMQMsg resultsMQMsg = JSON.parseObject(body, new TypeReference<SqlResultsMQMsg>() {
+                    });
+                    if (log.isDebugEnabled()) {
+                        log.debug("SqlResultMsgProcess Event Handler: {}", resultsMQMsg.getMsgId());
+                    }
+                    String resultDataSource = DataObject.getStringValue(resultsMQMsg.getResultDataSource(), Default_ResultDataSource);
+                    String resultTableName = DataObject.getStringValue(resultsMQMsg.getResultTableName(), Default_ResultTableName);
+                    List<ExecResult> saveDatas = resultsMQMsg.getSaveDatas();
+                    if (saveDatas != null && saveDatas.size() > 0) {
+                        List<List<ExecResult>> splist = LiteDataService.splitList(saveDatas, saveResultBatchSize);
+                        splist.forEach(array -> {
+                            execResultExService.saveResultBatch(array, resultDataSource, resultTableName);
+                        });
+                    }
+                    List<ExecResult> deleteDatas = resultsMQMsg.getDeleteDatas();
+                    if (deleteDatas != null && deleteDatas.size() > 0) {
+                        List<List<ExecResult>> splist = LiteDataService.splitList(deleteDatas, deleteResultBatchSize);
+                        splist.forEach(array -> {
+                            execResultExService.clearResultBatch(array, resultDataSource, resultTableName);
+                        });
+                    }
+                    //临时处理批次处理结果为10，表示数据插入完成
+                    ExecLog execlog = new ExecLog();
+                    execlog.setId(resultsMQMsg.getMsgId());
+                    execlog.setRetCode(10);
+                    ruExecLogService.updateById(execlog);
+                } else {
+                    ResultsMQMsg resultsMQMsg = JSON.parseObject(body, new TypeReference<ResultsMQMsg>() {});
+                    if(!StringUtils.isEmpty(resultsMQMsg.getKeyValueField())){
+                        execResultRepository.saveResultsMQMsg(resultsMQMsg);
+                    }
+                }
+            }
         } catch (Exception e) {
             log.error("获取MQ消息内容异常{}",e);
         }