feature: kafka消费

ruoyi-example/ruoyi-test-mq/pom.xml

@@ -68,7 +68,16 @@
                 </exclusion>
             </exclusions>
         </dependency>
-
+        <dependency>
+            <groupId>org.apache.dubbo</groupId>
+            <artifactId>dubbo</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.dromara</groupId>
+            <artifactId>ruoyi-api-backstage</artifactId>
+            <version>2.2.0</version>
+            <scope>compile</scope>
+        </dependency>
     </dependencies>
 
     <build>

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/config/KafkaStreamsConfig.java

@@ -11,11 +11,11 @@ import org.springframework.kafka.annotation.EnableKafkaStreams;
  *
  * @author LionLi
  */
-@Configuration
-@EnableKafkaStreams
+//@Configuration
+//@EnableKafkaStreams
 public class KafkaStreamsConfig {
 
-    @Bean
+    //@Bean
     public KStream<String, String> demoStream(StreamsBuilder builder) {
         // 输入主题
         KStream<String, String> source = builder.stream("input-topic");

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/consumer/KafkaNormalConsumer.java

@@ -1,23 +1,55 @@
 package org.dromara.stream.consumer;
 
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import lombok.AllArgsConstructor;
+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.RemoteConsumeService;
+import org.dromara.backstage.api.domain.bo.RemoteConsumptionBo;
+import org.dromara.common.core.domain.R;
+import org.dromara.common.core.enums.CreditTypeEnum;
+import org.dromara.stream.domain.bo.ConsumeRecordBo;
+import org.dromara.stream.domain.bo.ConsumptionBo;
+import org.dromara.stream.domain.bo.KafkaMessage;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
+import java.math.BigDecimal;
+import java.util.Date;
+
 /**
  * @author xbhog
  * @date 2024/05/19 18:04
  **/
 @Slf4j
 @Component
+@RequiredArgsConstructor
 public class KafkaNormalConsumer {
-
+    @DubboReference
+    private final RemoteConsumeService remoteConsumeService;
     //默认获取最后一条消息
-    @KafkaListener(topics = "test-topic", groupId = "test-group-id")
+    @KafkaListener(topics = "eventBus", groupId = "test-group-id")
     public void timiKafka(ConsumerRecord<String, String> record) {
         Object key = record.key();
         Object value = record.value();
+        KafkaMessage<?> kmsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
+        JSONObject data = JSONUtil.parseObj(kmsg.getBody());
+        RemoteConsumptionBo recordBo = new RemoteConsumptionBo();
+        recordBo.setConsumeMoney((BigDecimal) data.get("consumeValue"));
+        recordBo.setConsumeDate((Date) data.get("consumeDate"));
+        recordBo.setCardNo((Long) data.get("cardNo"));
+        recordBo.setFactoryId((Long) data.get("factoryFixId"));
+        recordBo.setTermNo((Long) data.get("termNo"));
+        recordBo.setTermRecordId((Long) data.get("termRecordID"));
+        recordBo.setRealName((String) data.get("xm"));
+        recordBo.setUserNumb((String) data.get("userNumb"));
+        recordBo.setRecordStatus((Long) data.get("posRecordState"));
+        recordBo.setCreditType(CreditTypeEnum.TERM_CONSUME);
+        R<Object> result = remoteConsumeService.receiveConsumeOriginalRecord(recordBo);
+
         log.info("【消费者】received the message key {}，value：{}", key, value);
     }
 

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/consumer/NormalRocketConsumer.java

@@ -11,7 +11,7 @@ import org.springframework.stereotype.Component;
  * @date 2024/06/01 16:53
  **/
 @Slf4j
-@Component
+//@Component
 @RocketMQMessageListener(topic = "test-topic", consumerGroup = "test-consumer-group")
 public class NormalRocketConsumer implements RocketMQListener<MessageExt> {
 

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/consumer/RabbitConsumer.java

@@ -12,7 +12,7 @@ import org.springframework.stereotype.Component;
  * @date 2024年5月18日
  */
 @Slf4j
-@Component
+//@Component
 public class RabbitConsumer {
 
     /**

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/consumer/TransactionRocketConsumer.java

@@ -10,7 +10,7 @@ import org.springframework.stereotype.Component;
  * @date 2024/06/01 16:54
  **/
 @Slf4j
-@Component
+//@Component
 @RocketMQMessageListener(topic = "transaction-topic", consumerGroup = "transaction-group")
 public class TransactionRocketConsumer  implements RocketMQListener<String> {
 

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/domain/bo/KafkaHeader.java

@@ -0,0 +1,34 @@
+package org.dromara.stream.domain.bo;
+
+import cn.hutool.core.util.IdUtil;
+import lombok.Data;
+
+/**
+ * name: KafkaHeader
+ * package: org.dromara.stream.domain.bo
+ * description: kafka消息头
+ * date: 2024-10-15 11:22:26 11:22
+ *
+ * @author luoyibo
+ * @version 0.1
+ * @since JDK 1.8
+ */
+@Data
+public class KafkaHeader {
+    /**
+     * 发送方
+     */
+    private String sender;
+    /**
+     * 时间戳
+     */
+    private Long timestamp;
+    /**
+     * 事件类型
+     */
+    private String eventType;
+    /**
+     * 事件Id
+     */
+    private String eventId;
+}

ruoyi-example/ruoyi-test-mq/src/main/java/org/dromara/stream/domain/bo/KafkaMessage.java

@@ -0,0 +1,20 @@
+package org.dromara.stream.domain.bo;
+
+import lombok.Data;
+
+/**
+ * name: KafkaMessage
+ * package: org.dromara.stream.domain.bo
+ * description: Kafka消息内容
+ * date: 2024-10-15 11:28:02 11:28
+ *
+ * @author luoyibo
+ * @version 0.1
+ * @since JDK 1.8
+ */
+@Data
+public class KafkaMessage<T> {
+    private KafkaHeader header=new KafkaHeader();
+
+    private T body;
+}

ruoyi-example/ruoyi-test-mq/src/main/resources/application.yml

@@ -23,7 +23,7 @@ spring:
 --- # kafka 配置
 spring:
   kafka:
-    bootstrap-servers: localhost:9092
+    bootstrap-servers: kafka04:18182,kafka05:18283,kafka06:18384
     consumer:
       group-id: test-group-id # 消费者组ID
       auto-offset-reset: earliest # 当没有偏移量或偏移量无效时，从何处开始消费

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/mq/KafkaNormalConsumer.java

@@ -44,13 +44,18 @@ public class KafkaNormalConsumer {
         KafkaMessage<?> kmsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
         if(kmsg.getHeader().getEventType().equals("00500001")){
             JSONObject data = JSONUtil.parseObj(kmsg.getBody());
-            //uploadByHttp(data);
-            uploadByService(data);
+            try {
+                //uploadByHttp(data);
+
+                uploadByService(data);
+            } catch (Exception e) {
+                log.info("消费失败");
+            }
         }
         log.info("【消费者】received the message key {}，value：{}", key, value);
     }
     private void uploadByService(JSONObject data) {
-        ConsumptionBo recordBo = new ConsumptionBo();
+                    ConsumptionBo recordBo = new ConsumptionBo();
         String time = data.get("consumeDate").toString();
         recordBo.setConsumeMoney(new BigDecimal(data.get("consumeValue").toString()));
         recordBo.setConsumeDate(DateUtil.parse(time));
@@ -70,6 +75,7 @@ public class KafkaNormalConsumer {
         recordBo.setConsumeId(vo.getOriginalId());
         consumeStrategyContent.postOrder(recordBo);
     }
+
     private void uploadByHttp(JSONObject data) {
         ConsumeRecordBo recordBo = new ConsumeRecordBo();
         String time = data.get("consumeDate").toString();