fix: 消费服务

1.消费定时对账

ruoyi-server/ruoyi-server-consume/src/main/java/org/dromara/server/consume/RuoYiServerConsumeApplication.java

@@ -5,6 +5,7 @@ import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.context.metrics.buffering.BufferingApplicationStartup;
 import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.annotation.EnableScheduling;
 
 /**
  * 系统模块
@@ -13,6 +14,7 @@ import org.springframework.scheduling.annotation.EnableAsync;
  */
 @EnableDubbo
 @EnableAsync
+@EnableScheduling
 @SpringBootApplication
 public class RuoYiServerConsumeApplication {
     public static void main(String[] args) {

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

@@ -269,7 +269,7 @@ public class CheckBusiness {
         if (R.isError(result)) {
             return result;
         }
-        // 5.可以消费，如果需要送kafka消息则异步发送，主要供云端重新走消费流程
+
         return R.ok();
     }
 

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

@@ -203,8 +203,9 @@ public class ConsumeBusiness {
         list.forEach(p->{
             p.setUseType(SystemUseTypeEnum.CONSUME.code());
             p.setCreditType(baseBusiness.getCreditType(p.getStatusFlag()));
+            p.setRecordStatus(364L);
             try {
-                R<ErrorInfo> result = fullOrder(p, "", "");
+                R<ErrorInfo> result = postOrder(p, "", "");
                 if (R.isSuccess(result)) {
                     doMessage.add(MessageFormat.format("[入账成功]-[{0}]", JsonUtils.toJsonString(p)));
                     success.getAndIncrement();
@@ -213,7 +214,7 @@ public class ConsumeBusiness {
                     fail.getAndIncrement();
                 }
             } catch (Exception e) {
-                doMessage.add(MessageFormat.format("[入账失败]-[{}]=[{}]", JsonUtils.toJsonString(p),e.getStackTrace()));
+                doMessage.add(MessageFormat.format("[入账失败]-[{0}]-[{1}]", JsonUtils.toJsonString(p), e.getStackTrace()));
                 fail.getAndIncrement();
                 log.error("[对账失败]-[{}]", JSONUtil.toJsonStr(p), e);
             }

ruoyi-server/ruoyi-server-consume/src/main/java/org/dromara/server/consume/task/ScheduledTasks.java

@@ -0,0 +1,39 @@
+package org.dromara.server.consume.task;
+
+import cn.hutool.core.date.DateUtil;
+import cn.hutool.json.JSONUtil;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.core.domain.R;
+import org.dromara.common.core.domain.model.ErrorInfo;
+import org.dromara.server.consume.business.ConsumeBusiness;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * name: ScheduledTasks
+ * package: org.dromara.backstage.task
+ * description: 定时任务
+ * date: 2024-12-11 13:32:36 13:32
+ *
+ * @author luoyibo
+ * @version 0.1
+ * @since JDK 1.8
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class ScheduledTasks {
+    private final ConsumeBusiness consumeBusiness;
+
+    @Scheduled(cron = "0 45 9,14,20,23 * * *")
+    public void syncDoorOpenRecord() {
+        String consumeDate = DateUtil.format(DateUtil.date(), "yyyy-MM-dd 00:00:00");
+        R<ErrorInfo> result = consumeBusiness.originalReconciliation(consumeDate);
+        if (R.isError(result)) {
+            log.error(JSONUtil.toJsonStr(result.getData()));
+        } else {
+            log.warn(result.getMsg());
+        }
+    }
+}

ruoyi-server/ruoyi-server-consume/src/main/resources/mapper/consume/ConsumeDetailOriginalMapper.xml

@@ -43,12 +43,14 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
 
     <select id="selectReconciliationData" resultType="org.dromara.server.common.domain.consume.bo.ConsumptionBo">
         SELECT term_no as termNo,user_numb as userNumb,card_no as cardNo,factory_id as factoryId
-             ,consume_date  as consumeDate,consume_money as consumeMoney,term_record_id as termRecordId,status_flag  as statusFlag
+        ,consume_date as consumeDate,consume_money as consumeMoney,term_record_id as termRecordId,status_flag as
+        statusFlag,
+        record_id as recordId
         FROM t_xf_consumedetailoriginal txc
         <where>
-            and original_id NOT IN (SELECT original_id FROM t_xf_consumedetail )
+            and original_id NOT IN (SELECT original_id FROM t_xf_consumedetail where consume_date>#{consumeDate})
             <if test="consumeDate != null">
-                AND txc.create_time>#{consumeDate}
+                AND txc.consume_date>#{consumeDate}
             </if>
         </where>
         ORDER BY consume_date ASC;

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

@@ -15,7 +15,6 @@ import org.dromara.common.message.kafka.constant.KafkaTopicConstants;
 import org.dromara.common.message.kafka.domain.KafkaHeader;
 import org.dromara.common.message.kafka.domain.KafkaMessage;
 import org.dromara.server.base.service.yktOperation.SyncRemoteSendMessageRecordService;
-import org.dromara.server.mq.domain.bo.XfOffsetBo;
 import org.dromara.server.mq.event.kafka.IYktEventStrategy;
 import org.dromara.server.mq.service.IXfOffsetService;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
@@ -44,6 +43,26 @@ public class KafkaCloudConsumer {
 
     private final IXfOffsetService xfOffsetService;
 
+    /**
+     * 初始化消息记录Bo
+     *
+     * @param consumeStatus 消费状态
+     * @param topic
+     */
+    private static RemoteSendMessageRecordBo initBo(KafkaMessage<?> msg, String consumeStatus, String topic) {
+        RemoteSendMessageRecordBo bo = new RemoteSendMessageRecordBo();
+        KafkaHeader header = msg.getHeader();
+        bo.setMessage("kafka");
+        bo.setTopic(topic);
+        bo.setEventId(header.getEventId());
+        bo.setSender(header.getSender());
+        bo.setEventType(header.getEventType());
+        bo.setMessage(msg.getBody().toString());
+        bo.setTenantId(header.getTenantId());
+        bo.setConsumeStatus(consumeStatus);
+        return bo;
+    }
+
     /**
      * eventBus主题监听 第三方对接相关
      *
@@ -71,7 +90,7 @@ public class KafkaCloudConsumer {
         String sender = receiveMsg.getHeader().getSender();
         //在eventBus主题中，sender=005是由本系统发出，无需业务处理
         if (ObjUtil.notEqual(sender, "005") && ObjUtil.notEqual(sender, "006")) {
-            doMessageHandle(receiveMsg);
+            doMessageHandle(receiveMsg, "");
         }
 
         // 记录
@@ -103,34 +122,19 @@ public class KafkaCloudConsumer {
         }
 
         KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
-        doMessageHandle(receiveMsg);
+        doMessageHandle(receiveMsg, topic);
 
         // 记录offset
         xfOffsetService.insert(IdUtil.simpleUUID(), offset, topic, groupId, value, new Date(),partition);
     }
 
-    /**
-     * 初始化消息记录Bo
-     *
-     * @param consumeStatus 消费状态
-     */
-    private static RemoteSendMessageRecordBo initBo(KafkaMessage<?> msg, String consumeStatus) {
-        RemoteSendMessageRecordBo bo = new RemoteSendMessageRecordBo();
-        KafkaHeader header = msg.getHeader();
-        bo.setEventId(header.getEventId());
-        bo.setSender(header.getSender());
-        bo.setEventType(header.getEventType());
-        bo.setMessage(msg.getBody().toString());
-        bo.setConsumeStatus(consumeStatus);
-        return bo;
-    }
-
     /**
      * 公用消息消费处理事件
      *
      * @param receiveMsg 接收到的消息信息
+     * @param topic 主题
      */
-    private void doMessageHandle(KafkaMessage<?> receiveMsg) {
+    private void doMessageHandle(KafkaMessage<?> receiveMsg, String topic) {
         try {
             JSONObject eventMsg = JSONUtil.parseObj(receiveMsg.getBody());
             String eventType = receiveMsg.getHeader().getEventType();
@@ -152,14 +156,14 @@ public class KafkaCloudConsumer {
             IYktEventStrategy eventStrategy = SpringUtils.getBean(sender, IYktEventStrategy.class);
             eventStrategy.doMsgHandle(eventType, eventMsg);
             try {
-                syncRemoteSendMessageRecordService.insertSendMessageRecord(initBo(receiveMsg, "Y"));
+                syncRemoteSendMessageRecordService.insertSendMessageRecord(initBo(receiveMsg, "Y", topic));
             } catch (Exception ex) {
                 log.error("消息发送记录更新失败", ex);
             }
 
         } catch (Exception e) {
             try {
-                syncRemoteSendMessageRecordService.insertSendMessageRecord(initBo(receiveMsg, "N"));
+                syncRemoteSendMessageRecordService.insertSendMessageRecord(initBo(receiveMsg, "N", topic));
             } catch (Exception ex) {
                 log.error("消息发送记录更新失败", ex);
             }