update:代码完善

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

@@ -104,11 +104,11 @@ 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));
+        log.trace(String.format("sendSqlResultsMsg:%1$s", msg));
         dstRocketMQTemplate.asyncSendOrderly(resultsTopic+":SqlResult",msg, DigestUtils.md5DigestAsHex(msg.getBytes()),new SendCallback(){
             @Override
             public void onSuccess(SendResult sendResult) {
-                log.info("结果消息发送响应：" + sendResult.toString());
+                log.info("SqlResultsMsg结果消息发送响应：" + sendResult.toString());
             }
 
             @Override

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

@@ -40,9 +40,6 @@ 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;
 
@@ -74,43 +71,42 @@ public class ResultMQMsgListener implements RocketMQListener<MessageExt> {
             String tags = messageExt.getTags();
             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 (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);
                     });
-                    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);
-                    }
+                }
+                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);
+            log.error("获取MQ消息内容异常{}", e);
         }
     }