rocketMQ优化

SLN/%PUBPRJ%-core/src/main/java/%SYS_PKGPATH%/core/util/config/RocketMQConfig.java.ftl

@@ -29,6 +29,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
 
 @Slf4j
 @Configuration
@@ -109,7 +110,7 @@ public class RocketMQConfig {
          */
 //        consumer.setMessageModel(MessageModel.CLUSTERING);
         try {
-            consumer.subscribe("default", "*");
+            consumer.subscribe("default", "${getSubscribeTags()}");
             consumer.start();
             log.info("rocketmq consumer 创建成功 groupName={}, topics={}, namesrvAddr={}", groupName, "default", namesrvAddr);
         } catch (MQClientException e) {
@@ -119,4 +120,21 @@ public class RocketMQConfig {
     }
     </#if>
 }
-</#if>
+</#if>
\ No newline at end of file
+
+<#function getSubscribeTags>
+    <#assign result="">
+    <#list sys.getAllPSDataEntities() as dataEntity>
+        <#if dataEntity.getAllPSDEDataSyncs?? && dataEntity.getAllPSDEDataSyncs()??>
+            <#list dataEntity.getAllPSDEDataSyncs() as dataSync>
+                <#if dataSync.getInPSSysDataSyncAgent?? && dataSync.getInPSSysDataSyncAgent()??>
+                    <#if result!="">
+                        <#assign result=result+" || ">
+                    </#if>
+                    <#assign result=result+dataSync.codeName?lower_case>
+                </#if>
+            </#list>
+        </#if>
+    </#list>
+    <#return result>
+</#function>
\ No newline at end of file

SLN/%PUBPRJ%-core/src/main/java/%SYS_PKGPATH%/core/util/config/RocketMQListenerProcessor.java.ftl

@@ -69,7 +69,7 @@ public class RocketMQListenerProcessor implements MessageListenerOrderly {
                             <#assign dataSyncCodeName=dataSync.codeName?lower_case>
                             <#if dataSync.getInTestPSDEAction?? && dataSync.getInTestPSDEAction()??>
                                 <#assign inputAction=srfmethodname(dataSync.getInTestPSDEAction().codeName)>
-        if ("${dataSyncCodeName}".equalsIgnoreCase(messageExt.getTopic())) {
+        if ("${dataSyncCodeName}".equalsIgnoreCase(tags)) {
             ${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName} domain = JSON.parseObject(new String(body),${pub.getPKGCodeName()}.core.${dataEntity.getPSSystemModule().getCodeName()?lower_case}.domain.${entityName}.class);
             ${entityName}Service.${inputAction}(domain);
         }