feature: 消费服务完善

1.增加了开关参数控制是否根据kafka消费数据云平台重新消费
2.增加了开关参数控制是否kafka是否消费原始消费请求

pom.xml

@@ -92,12 +92,12 @@
             <id>prod</id>
             <properties>
                 <profiles.active>prod</profiles.active>
-                <nacos.server>10.32.23.157:8848</nacos.server>
+                <nacos.server>172.16.137.72:8848</nacos.server>
                 <nacos.discovery.group>DEFAULT_GROUP</nacos.discovery.group>
                 <nacos.config.group>DEFAULT_GROUP</nacos.config.group>
                 <nacos.username>nacos</nacos.username>
                 <nacos.password>nacos</nacos.password>
-                <logstash.address>10.32.23.157:4560</logstash.address>
+                <logstash.address>172.16.137.72:4560</logstash.address>
             </properties>
             <activation>
                 <!-- 默认环境 -->

ruoyi-server/ruoyi-server-consume/src/main/java/org/dromara/server/consume/business/ConsumeBusiness.java

@@ -56,12 +56,12 @@ public class ConsumeBusiness {
             return result;
         }
 
-        log.info("[请求交易]-[mac校验]-[{}]", JSONUtil.toJsonStr(bo));
-        result = checkBusiness.checkMac(bo, mac);
-        if (R.isError(result)) {
-            log.error("[请求交易]-[mac校验失败]-[{}]", JSONUtil.toJsonStr(result.getData()));
-            return result;
-        }
+        //log.info("[请求交易]-[mac校验]-[{}]", JSONUtil.toJsonStr(bo));
+        //result = checkBusiness.checkMac(bo, mac);
+        //if (R.isError(result)) {
+        //    log.error("[请求交易]-[mac校验失败]-[{}]", JSONUtil.toJsonStr(result.getData()));
+        //    return result;
+        //}
 
         log.info("[请求交易]-[用户信息验证]-[{}]", JSONUtil.toJsonStr(bo));
         RemoteUserAccountVo userAccountVo = new RemoteUserAccountVo();

ruoyi-server/ruoyi-server-consume/src/main/java/org/dromara/server/consume/controller/v1/ConsumeController.java

@@ -144,7 +144,8 @@ public class ConsumeController {
         R<ErrorInfo> errorInfo;
         if (Objects.equals(type, "requestConsume")) {
             errorInfo = consumeBusiness.createOrder(bo, mac, xfPwd);
-            String cloudConsume = remotePtParameterService.getPtParameterByKey("CLOUD_CONSUME");
+            //是否推送云端消费
+            String cloudConsume = remotePtParameterService.getPtParameterByKey("PUSH_CLOUD_CONSUME");
             if(ObjectUtil.equals(cloudConsume,"1")){
                 baseBusiness.sendCloudConsume(cloudConsumeBo);
             }

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/consumer/KafkaConsumer.java

@@ -1,13 +1,17 @@
 package org.dromara.server.mq.consumer;
 
 import cn.hutool.core.util.ObjUtil;
+import cn.hutool.core.util.ObjectUtil;
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.dubbo.config.annotation.DubboReference;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.dromara.backstage.api.RemotePtParameterService;
 import org.dromara.common.message.kafka.domain.KafkaMessage;
 import org.dromara.server.mq.event.kafka.EventStrategyContext;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
@@ -24,19 +28,26 @@ import org.springframework.stereotype.Component;
 @RequiredArgsConstructor
 @Slf4j
 @Component
+@ConditionalOnExpression("'cloud'.equals('${locationFlag}')")
 public class KafkaConsumer {
     private final EventStrategyContext eventStrategyContext;
+    @DubboReference
+    private final RemotePtParameterService remotePtParameterService;
 
-    @KafkaListener(topics = "TO_CLOUD_EVENT", groupId = "test-group-id")
+    @KafkaListener(topics = "TO_CLOUD_EVENT", groupId = "test1-group-id")
     public void kafkaReceiveHandler(ConsumerRecord<String, String> record) {
         KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
-        //log.info("[接收到Kafka消息]-[{}]", receiveMsg);
         String eventType = receiveMsg.getHeader().getEventType();
         String sender = receiveMsg.getHeader().getSender();
         JSONObject eventMsg = JSONUtil.parseObj(receiveMsg.getBody());
         if(ObjUtil.equals(sender,"000")) {
             try {
-                eventStrategyContext.doMsgHandle(eventType, eventMsg);
+                //是接收云端消费
+                String cloudConsume = remotePtParameterService.getPtParameterByKey("GET_CLOUD_CONSUME");
+                if(ObjectUtil.equals(cloudConsume,"1")){
+                    log.info("[处理云端消费请求]-[消信息:{}]", receiveMsg);
+                    eventStrategyContext.doMsgHandle(eventType, eventMsg);
+                }
             } catch (Exception e) {
                 log.error("[kafka消息处理失败]-[消息:{}-[错误:{}]", receiveMsg, e.getMessage());
             }