feature: 根据参数是否启用来发送请求云端消费和推送消费记录

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/domain/vo/RemoteUserAccountVo.java

@@ -143,4 +143,9 @@ public class RemoteUserAccountVo implements Serializable {
      * 账户发卡信息
      */
     private String accountCard;
+
+    /**
+     * 部门唯一身份标识,第三方统一身份认证ID
+     */
+    private String otherDeptId;
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/business/self/TraineeBusiness.java

@@ -402,7 +402,8 @@ public class TraineeBusiness {
         KafkaHeader header = message.getHeader();
         header.setTimestamp(System.currentTimeMillis());
         header.setEventId(YktEventConstaints.CHECKIN_STATE);
-        header.setEventType(YktEventConstaints.SENDER);
+        header.setEventType(YktEventConstaints.CHECKIN_STATE);
+        header.setSender(YktEventConstaints.SENDER);
         message.setHeader(header);
         message.setBody(traineeCheckIn);
 

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/payment/domain/vo/PtUserAccountVo.java

@@ -1,19 +1,14 @@
 package org.dromara.backstage.payment.domain.vo;
 
-import java.util.Date;
-import com.fasterxml.jackson.annotation.JsonFormat;
-import org.dromara.backstage.payment.domain.PtUserAccount;
 import com.alibaba.excel.annotation.ExcelIgnoreUnannotated;
 import com.alibaba.excel.annotation.ExcelProperty;
-import org.dromara.common.excel.annotation.ExcelDictFormat;
-import org.dromara.common.excel.convert.ExcelDictConvert;
 import io.github.linpeilie.annotations.AutoMapper;
 import lombok.Data;
-import org.dromara.common.sensitive.annotation.Sensitive;
-import org.dromara.common.sensitive.core.SensitiveStrategy;
+import org.dromara.backstage.payment.domain.PtUserAccount;
+import org.dromara.common.excel.annotation.ExcelDictFormat;
+import org.dromara.common.excel.convert.ExcelDictConvert;
 import org.dromara.common.translation.annotation.Translation;
 import org.dromara.common.translation.constant.TransConstant;
-import org.springframework.boot.autoconfigure.web.WebProperties;
 
 import java.io.Serial;
 import java.io.Serializable;
@@ -241,4 +236,9 @@ public class PtUserAccountVo implements Serializable {
      */
     @Translation(type = TransConstant.USER_ID_TO_CARD_INFO, mapper = "userId")
     private String accountCard;
+
+    /**
+     * 部门唯一身份标识,第三方统一身份认证ID
+     */
+    private String otherDeptId;
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/payment/service/impl/PtUserAccountServiceImpl.java

@@ -38,6 +38,7 @@ import org.dromara.common.mybatis.core.page.PageQuery;
 import org.dromara.common.mybatis.core.page.TableDataInfo;
 import org.dromara.common.redis.utils.RedisUtils;
 import org.dromara.system.api.RemoteDeptService;
+import org.dromara.system.api.domain.vo.RemoteDeptVo;
 import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 
@@ -79,8 +80,9 @@ public class PtUserAccountServiceImpl implements IPtUserAccountService {
     public PtUserAccountVo queryById(Long userId) {
         PtUserAccountVo vo = baseMapper.selectVoById(userId);
         if (vo != null) {
-            String deptName = remoteDeptService.selectDeptNameByIds(vo.getDeptId().toString());
-            vo.setDeptName(deptName);
+            RemoteDeptVo remoteDeptVo = remoteDeptService.selectDeptById(vo.getDeptId());
+            vo.setDeptName(remoteDeptVo.getDeptName());
+            vo.setOtherDeptId(remoteDeptVo.getOtherId());
         }
         return vo;
     }

ruoyi-server/ruoyi-server-base/pom.xml

@@ -53,5 +53,13 @@
             <version>2.2.0</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
     </dependencies>
 </project>

ruoyi-server/ruoyi-server-base/src/main/java/org/dromara/server/base/mq/KafkaNormalProducer.java

@@ -0,0 +1,38 @@
+package org.dromara.server.base.mq;
+
+import com.alibaba.fastjson2.JSON;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.dromara.common.message.kafka.domain.KafkaMessage;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.CompletableFuture;
+
+@RequiredArgsConstructor
+@Slf4j
+@Component
+public class KafkaNormalProducer {
+    private final KafkaTemplate<String, String> kafkaTemplate;
+
+    public void sendKafkaMessage(String topic, KafkaMessage<?> data) {
+        String jsonMessage = JSON.toJSONString(data);
+        try {
+            ProducerRecord<String, String> record;
+            record = new ProducerRecord<>(topic, jsonMessage);
+
+            CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(record);
+            send.whenComplete((result, ex) -> {
+                if (ex != null) {
+                    log.error("消费系统发送到kafka消息系统异常，data: {}", jsonMessage, ex);
+                } else {
+                    log.info("消费系统发送到kafka消息系统成功，data: {}", jsonMessage);
+                }
+            });
+        } catch (Exception e) {
+            log.error("消费系统发送到kafka消息系统异常，data: {}", data, e);
+        }
+    }
+}

ruoyi-server/ruoyi-server-base/src/main/java/org/dromara/server/base/mq/constant/CloudConsumeEventConstants.java

@@ -0,0 +1,17 @@
+package org.dromara.server.base.mq.constant;
+
+/**
+ * @ClassName PushConsumeEventConstants
+ * @Description 云端消费请求事件常量
+ * @Author luoyibo
+ * @Date 2024-12-27 22:11
+ * @Version 1.0
+ * @since jdk17
+ */
+public class CloudConsumeEventConstants {
+    // 业务系统编号
+    public static final String SENDER = "000";
+
+    // 消费记录
+    public static final String CONSUME_DETAIL = SENDER + "00001";
+}

ruoyi-server/ruoyi-server-base/src/main/java/org/dromara/server/base/mq/constant/PushConsumeEventConstants.java

@@ -0,0 +1,17 @@
+package org.dromara.server.base.mq.constant;
+
+/**
+ * @ClassName PushConsumeEventConstants
+ * @Description 消费记录推送事件常量
+ * @Author luoyibo
+ * @Date 2024-12-27 22:11
+ * @Version 1.0
+ * @since jdk17
+ */
+public class PushConsumeEventConstants {
+    // 业务系统编号
+    public static final String SENDER = "005";
+
+    // 消费记录
+    public static final String CONSUME_DETAIL = SENDER + "00001";
+}

ruoyi-server/ruoyi-server-base/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -21,3 +21,4 @@ org.dromara.server.base.service.yktOperation.SyncRemoteXfQuotaService
 org.dromara.server.base.service.yktOperation.SyncRemoteXfLimitedService
 org.dromara.server.base.service.yktOperation.SyncRemoteSendMessageRecordService
 org.dromara.server.base.service.yktOperation.SyncRemotePtSubsidyService
+org.dromara.server.base.mq.KafkaNormalProducer

ruoyi-server/ruoyi-server-common/src/main/java/org/dromara/server/common/domain/vo/yc/YcPushConsumeInfoVo.java

@@ -0,0 +1,62 @@
+package org.dromara.server.common.domain.vo.yc;
+
+import lombok.Data;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+/**
+ * 需要推送的消费信息
+ *
+ * @author LionLi
+ * @date 2024-08-06
+ */
+@Data
+public class YcPushConsumeInfoVo implements Serializable {
+    @Serial
+    private static final long serialVersionUID = -879441600292791268L;
+    // 记录id
+	private String recordId;
+	// 人员id
+	private String userId;
+	// 学号
+	private String userNumb;
+	// 姓名
+	private String xm;
+	// 部门id
+	private String deptId;
+	// 部门名称
+	private String deptName;
+	// 消费地点id
+	private String roomId;
+	// 地点名称
+	private String roomName;
+	// 卡流水号
+	private String cardNo;
+	// 物理卡号
+	private String factoryFixId;
+	// 消费金额
+	private String consumeValue;
+	// 卡余
+	private String cardValue;
+	// 消费时间
+	private String consumeDate;
+	// 餐类名称id
+	private String mealTypeId;
+	// 餐类名称
+	private String mealName;
+	// 机号
+	private String termNo;
+	// 机器名称
+	private String termName;
+	// 身份 0-系统内置 1-教师 2-学生 3-家长
+	private String category;
+	// 其他业务系统人员id,对应教务或者人事的人员id
+	private String otherSysId;
+	// 班级id
+	private String classId;
+	//机器流水号
+	private Long termRecordID;
+	//消费记录标识
+	private Integer posRecordState;
+}

ruoyi-server/ruoyi-server-consume/pom.xml

@@ -114,6 +114,14 @@
             <version>2.2.0</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

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

@@ -4,6 +4,7 @@ import org.apache.dubbo.config.spring.context.annotation.EnableDubbo;
 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;
 
 /**
  * 系统模块
@@ -11,6 +12,7 @@ import org.springframework.boot.context.metrics.buffering.BufferingApplicationSt
  * @author ruoyi
  */
 @EnableDubbo
+@EnableAsync
 @SpringBootApplication
 public class RuoYiServerConsumeApplication {
     public static void main(String[] args) {

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

@@ -3,8 +3,12 @@ package org.dromara.server.consume.business;
 import cn.hutool.core.bean.BeanUtil;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.core.util.ObjUtil;
+import cn.hutool.core.util.ObjectUtil;
+import cn.hutool.json.JSONUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.dubbo.config.annotation.DubboReference;
+import org.dromara.backstage.api.RemotePtParameterService;
 import org.dromara.backstage.api.domain.vo.RemoteCardVo;
 import org.dromara.backstage.api.domain.vo.RemoteMealTypeVo;
 import org.dromara.backstage.api.domain.vo.RemoteUserAccountVo;
@@ -16,10 +20,17 @@ import org.dromara.common.core.enums.BalanceUpdateEnum;
 import org.dromara.common.core.enums.ResultCodeEnum;
 import org.dromara.common.core.exception.consume.ConsumeException;
 import org.dromara.common.core.utils.RecordIdUtils;
+import org.dromara.common.message.kafka.domain.KafkaHeader;
+import org.dromara.common.message.kafka.domain.KafkaMessage;
+import org.dromara.server.base.mq.constant.CloudConsumeEventConstants;
 import org.dromara.server.common.domain.consume.bo.ConsumptionBo;
+import org.dromara.server.common.domain.vo.yc.YcPushConsumeInfoVo;
+import org.dromara.server.base.mq.KafkaNormalProducer;
+import org.dromara.server.base.mq.constant.PushConsumeEventConstants;
 import org.dromara.server.consume.domain.bo.*;
 import org.dromara.server.consume.domain.vo.*;
 import org.dromara.server.consume.service.*;
+import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -48,6 +59,10 @@ public class BaseBusiness {
     private final IXfTermTotalService termTotalService;
     private final IXfConsumeDetailService consumeDetailService;
     private final IPtBagService bagService;
+    private final KafkaNormalProducer kafkaProducer;
+
+    @DubboReference
+    private final RemotePtParameterService remotePtParameterService;
 
     /**
      * 生成原始消费记录
@@ -143,6 +158,12 @@ public class BaseBusiness {
         if (!updateBagBalance(bagVos)) {
             return R.fail(new ErrorInfo(400, ApiErrorTypeConstants.EXCEPTION, "更新钱包余额表失败", ""));
         }
+
+        // 5.发送一条消费记录到kafka
+        String pushConsume = remotePtParameterService.getPtParameterByKey("PUSH_CONSUME_MQ");
+        if(ObjectUtil.equals(pushConsume, "1")){
+            this.sendConsumeToKafka(detailVos,userAccountVo);
+        }
         return R.ok();
     }
 
@@ -222,6 +243,22 @@ public class BaseBusiness {
         return result.get();
     }
 
+    @Async
+    public void sendCloudConsume(ConsumptionBo bo){
+        KafkaMessage<ConsumptionBo> message = new KafkaMessage<>();
+        KafkaHeader header = message.getHeader();
+        header.setTimestamp(System.currentTimeMillis());
+        header.setEventId(CloudConsumeEventConstants.CONSUME_DETAIL);
+        header.setEventType(CloudConsumeEventConstants.CONSUME_DETAIL);
+        header.setSender(CloudConsumeEventConstants.SENDER);
+
+        message.setHeader(header);
+        message.setBody(bo);
+
+        kafkaProducer.sendKafkaMessage("TO_CLOUD_EVENT", message);
+        log.info("请求云端消费:{}", JSONUtil.toJsonStr(bo));
+    }
+
     /**
      * 创建一条消费明细记录
      *
@@ -274,5 +311,50 @@ public class BaseBusiness {
         return consumeDetailService.createConsumeDetailRecord(consumeDetailBo);
     }
 
+    /**
+     * 将消费信息发送到kafka
+     * @param consumeList 消费记录列表
+     * @param accountVo 消费人员信息
+     */
+    private void sendConsumeToKafka(List<XfConsumeDetailVo> consumeList,RemoteUserAccountVo accountVo){
+        for (XfConsumeDetailVo vo : consumeList) {
+            YcPushConsumeInfoVo ycSendConsumeInfo = new YcPushConsumeInfoVo();
+            ycSendConsumeInfo.setRecordId(vo.getRecordId().toString());
+            ycSendConsumeInfo.setUserId(vo.getUserId().toString());
+            ycSendConsumeInfo.setUserNumb(vo.getUserNumb());
+            ycSendConsumeInfo.setXm(vo.getRealName());
+            ycSendConsumeInfo.setDeptId(vo.getDeptId().toString());
+            ycSendConsumeInfo.setDeptName(vo.getDeptName());
+            ycSendConsumeInfo.setRoomId(vo.getRoomId().toString());
+            ycSendConsumeInfo.setRoomName(vo.getRoomName());
+            ycSendConsumeInfo.setCardNo(vo.getCardNo().toString());
+            ycSendConsumeInfo.setFactoryFixId(vo.getFactoryId().toString());
+            ycSendConsumeInfo.setConsumeValue(vo.getConsumeMoney().toString());
+            ycSendConsumeInfo.setCardValue(DateUtil.format(vo.getConsumeDate(),DefaultConstants.DATE_TIME_FORMAT));
+            ycSendConsumeInfo.setConsumeDate(vo.getConsumeDate().toString());
+            ycSendConsumeInfo.setMealTypeId(vo.getMealType().toString());
+            ycSendConsumeInfo.setMealName(vo.getMealName());
+            ycSendConsumeInfo.setTermNo(vo.getTermNo().toString());
+            ycSendConsumeInfo.setTermName(vo.getTermName());
+            ycSendConsumeInfo.setCategory(accountVo.getCategory());
+            ycSendConsumeInfo.setOtherSysId(accountVo.getOtherId());
+            ycSendConsumeInfo.setClassId(accountVo.getOtherDeptId());
+            ycSendConsumeInfo.setTermRecordID(vo.getTermRecordId());
+            ycSendConsumeInfo.setPosRecordState(vo.getRecordStatus().intValue());
+
+            KafkaMessage<YcPushConsumeInfoVo> message = new KafkaMessage<>();
+            KafkaHeader header = message.getHeader();
+            header.setTimestamp(System.currentTimeMillis());
+            header.setEventId(PushConsumeEventConstants.CONSUME_DETAIL);
+            header.setEventType(PushConsumeEventConstants.CONSUME_DETAIL);
+            header.setSender(PushConsumeEventConstants.SENDER);
+            message.setHeader(header);
+            message.setBody(ycSendConsumeInfo);
+
+            kafkaProducer.sendKafkaMessage("eventBus", message);
+            log.info("消费记录发送Kafka:{}", JSONUtil.toJsonStr(ycSendConsumeInfo));
+        }
+    }
+
 
 }

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

@@ -186,6 +186,8 @@ public class CheckBusiness {
     public R<ErrorInfo> checkConsume(ConsumptionBo bo, RemoteUserAccountVo userAccountVo, RemoteCardVo userCardVo, XfTermVo useTermVo) {
         ErrorInfo errorInfo;
         R<ErrorInfo> result;
+        // 如果折扣验证，消费金额会更新成折扣金额，所以先保存
+        BigDecimal consumeMoney = bo.getConsumeMoney();
         // 1.消费账户状态验证，验证账户是否已开户、是否冻结、状态是否正常、是否过有效期
         if ("Y".equals(userAccountVo.getFreezeStatus())) {
             errorInfo = new ErrorInfo(400, ApiErrorTypeConstants.NOT_FOUND, "账户已被冻结",
@@ -223,6 +225,7 @@ public class CheckBusiness {
         if (R.isError(result)) {
             return result;
         }
+        // 5.可以消费，如果需要送kafka消息则异步发送，主要供云端重新走消费流程
         return R.ok();
     }
 
@@ -696,34 +699,34 @@ public class CheckBusiness {
             0L) > 0 || fourCount.compareTo(0L) > 0) {
             if (dayCount.compareTo(0L) > 0) {
                 // 日限次
-                if (cardLimitedVo.getDayCount().compareTo(dayCount) > 0) {
+                if (cardLimitedVo.getDayCount().compareTo(dayCount) >= 0) {
                     return R.fail(
                         new ErrorInfo(400, TradeStatusEnum.DayLimitTimes.toString(), "卡类日限制次数", TradeStatusEnum.DayLimitTimes.getName()));
                 }
                 // 早餐限次
                 if (ObjectUtil.equals(mealType, "1")) {
-                    if (cardLimitedVo.getMealCount().compareTo(oneCount) > 0) {
+                    if (cardLimitedVo.getMealCount().compareTo(oneCount) >= 0) {
                         return R.fail(new ErrorInfo(400, TradeStatusEnum.MealLimitTimes.toString(), "卡类早餐限制次数",
                                                     TradeStatusEnum.MealLimitTimes.getName()));
                     }
                 }
                 // 午餐限次
                 if (ObjectUtil.equals(mealType, "2")) {
-                    if (cardLimitedVo.getMealCount().compareTo(twoCount) > 0) {
+                    if (cardLimitedVo.getMealCount().compareTo(twoCount) >= 0) {
                         return R.fail(new ErrorInfo(400, TradeStatusEnum.MealLimitTimes.toString(), "卡类午餐限制次数",
                                                     TradeStatusEnum.MealLimitTimes.getName()));
                     }
                 }
                 // 晚餐限次
                 if (ObjectUtil.equals(mealType, "3")) {
-                    if (cardLimitedVo.getMealCount().compareTo(threeCount) > 0) {
+                    if (cardLimitedVo.getMealCount().compareTo(threeCount) >= 0) {
                         return R.fail(new ErrorInfo(400, TradeStatusEnum.MealLimitTimes.toString(), "卡类晚餐限制次数",
                                                     TradeStatusEnum.MealLimitTimes.getName()));
                     }
                 }
                 // 宵夜限次
                 if (ObjectUtil.equals(mealType, "4")) {
-                    if (cardLimitedVo.getMealCount().compareTo(fourCount) > 0) {
+                    if (cardLimitedVo.getMealCount().compareTo(fourCount) >= 0) {
                         return R.fail(new ErrorInfo(400, TradeStatusEnum.MealLimitTimes.toString(), "卡类夜宵限制次数",
                                                     TradeStatusEnum.MealLimitTimes.getName()));
                     }

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

@@ -7,6 +7,7 @@ import org.dromara.backstage.api.domain.vo.RemoteCardVo;
 import org.dromara.backstage.api.domain.vo.RemoteMealTypeVo;
 import org.dromara.backstage.api.domain.vo.RemoteOperatorVo;
 import org.dromara.backstage.api.domain.vo.RemoteUserAccountVo;
+import org.dromara.common.core.config.AsyncConfig;
 import org.dromara.common.core.domain.R;
 import org.dromara.common.core.domain.model.ErrorInfo;
 import org.dromara.server.common.domain.consume.bo.ConsumptionBo;
@@ -34,6 +35,8 @@ import java.util.List;
 public class ConsumeBusiness {
     private final CheckBusiness checkBusiness;
     private final BaseBusiness baseBusiness;
+    private final AsyncConfig asyncConfig;
+    // private final KafkaNormalProducer kafkaProducer;
 
     /**
      * 请求消费
@@ -44,6 +47,8 @@ public class ConsumeBusiness {
      * @return 请求结果
      */
     public R<ErrorInfo> createOrder(ConsumptionBo bo, String mac, String xfPwd) {
+        // baseBusiness.sendCloudConsume(bo);
+
         log.info("[请求交易]-[开始参数验证]-[{}]", JSONUtil.toJsonStr(bo));
         R<ErrorInfo> result = checkBusiness.checkParam(bo);
         if (R.isError(result)) {
@@ -157,4 +162,17 @@ public class ConsumeBusiness {
         }
         return R.ok();
     }
+    // @Async
+    // protected void sendCloudConsume(ConsumptionBo bo){
+    //     KafkaMessage<ConsumptionBo> message = new KafkaMessage<>();
+    //     KafkaHeader header = message.getHeader();
+    //     header.setTimestamp(System.currentTimeMillis());
+    //     header.setEventId(CloudConsumeEventConstants.CONSUME_DETAIL);
+    //     header.setEventType(CloudConsumeEventConstants.SENDER);
+    //     message.setHeader(header);
+    //     message.setBody(bo);
+    //
+    //     // kafkaProducer.sendKafkaMessage("TO_CLOUD_EVENT", message);
+    //     log.info("请求云端消费:{}", JSONUtil.toJsonStr(bo));
+    // }
 }

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

@@ -1,9 +1,13 @@
 package org.dromara.server.consume.controller.v1;
 
+import cn.hutool.core.bean.BeanUtil;
 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.dromara.backstage.api.RemotePtParameterService;
 import org.dromara.common.core.constant.CacheNames;
 import org.dromara.common.core.domain.R;
 import org.dromara.common.core.domain.model.ErrorInfo;
@@ -12,6 +16,7 @@ import org.dromara.common.core.enums.CreditTypeEnum;
 import org.dromara.common.core.enums.SystemUseTypeEnum;
 import org.dromara.common.redis.utils.RedisUtils;
 import org.dromara.server.common.domain.consume.bo.ConsumptionBo;
+import org.dromara.server.consume.business.BaseBusiness;
 import org.dromara.server.consume.business.ConsumeBusiness;
 import org.dromara.server.consume.convert.strategy.RecordConvertStrategyContent;
 import org.springframework.http.HttpStatus;
@@ -32,6 +37,7 @@ import java.util.Objects;
  * @since JDK 1.8
  */
 
+@Slf4j
 @Validated
 @RestController
 @RequiredArgsConstructor
@@ -39,6 +45,9 @@ import java.util.Objects;
 public class ConsumeController {
     private final RecordConvertStrategyContent recordConvertStrategy;
     private final ConsumeBusiness consumeBusiness;
+    private final BaseBusiness baseBusiness;
+    @DubboReference
+    private final RemotePtParameterService remotePtParameterService;
 
     /**
      * 请求消费(校园码)
@@ -131,9 +140,14 @@ public class ConsumeController {
         bo.setUseType(SystemUseTypeEnum.CONSUME);
         bo.setCreditType(CreditTypeEnum.TERM_CONSUME);
 
+        ConsumptionBo cloudConsumeBo = BeanUtil.copyProperties(bo, ConsumptionBo.class);
         R<ErrorInfo> errorInfo;
         if (Objects.equals(type, "requestConsume")) {
             errorInfo = consumeBusiness.createOrder(bo, mac, xfPwd);
+            String cloudConsume = remotePtParameterService.getPtParameterByKey("CLOUD_CONSUME");
+            if(ObjectUtil.equals(cloudConsume,"1")){
+                baseBusiness.sendCloudConsume(cloudConsumeBo);
+            }
         } else if (Objects.equals(type, "uploadRecord")) {
             errorInfo = consumeBusiness.postOrder(bo, mac, xfPwd);
         } else {

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

@@ -6,17 +6,11 @@ import cn.hutool.json.JSONUtil;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.dromara.backstage.api.domain.bo.RemoteSendMessageRecordBo;
 import org.dromara.common.message.kafka.domain.KafkaMessage;
-import org.dromara.server.base.service.yktOperation.SyncRemoteSendMessageRecordService;
 import org.dromara.server.mq.event.kafka.EventStrategyContext;
-import org.dromara.server.mq.event.kafka.YktEventStrategyContext;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.stereotype.Component;
 
-import static org.dromara.common.message.kafka.constant.KafkaTopicConstants.SYNC_DATA_TOPIC;
-
 /**
  * name: KafkaNormalConsumer
  * package: org.dromara.server.mq.consumer
@@ -33,14 +27,14 @@ import static org.dromara.common.message.kafka.constant.KafkaTopicConstants.SYNC
 public class KafkaConsumer {
     private final EventStrategyContext eventStrategyContext;
 
-    @KafkaListener(topics = "eventBus", groupId = "test-group-id")
+    @KafkaListener(topics = "TO_CLOUD_EVENT", groupId = "test-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,"005")) {
+        if(ObjUtil.equals(sender,"000")) {
             try {
                 eventStrategyContext.doMsgHandle(eventType, eventMsg);
             } catch (Exception e) {

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/event/kafka/impl/card/CloudConsumeEventStrategyImpl.java

@@ -0,0 +1,52 @@
+package org.dromara.server.mq.event.kafka.impl.card;
+
+import cn.hutool.core.bean.BeanUtil;
+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.dromara.consume.api.RemoteConsumeService;
+import org.dromara.consume.api.domain.bo.RemoteConsumeBo;
+import org.dromara.server.base.mq.constant.CloudConsumeEventConstants;
+import org.dromara.server.common.domain.consume.bo.ConsumptionBo;
+import org.dromara.server.mq.event.kafka.IEventStrategy;
+import org.springframework.stereotype.Service;
+
+/**
+ * @ClassName ConsumeEventStrategyImpl
+ * @Description TODO
+ * @Author luoyibo
+ * @Date 2024-11-12 11:52
+ * @Version 1.0
+ * @since jdk17
+ */
+@Slf4j
+@RequiredArgsConstructor
+@Service(CloudConsumeEventConstants.CONSUME_DETAIL)
+public class CloudConsumeEventStrategyImpl implements IEventStrategy {
+    @DubboReference
+    private final RemoteConsumeService remoteConsumeService;
+
+    @Override
+    public void doMsgHandle(JSONObject data) {
+        ConsumptionBo consumptionBo = JSONUtil.toBean(data, ConsumptionBo.class);
+        RemoteConsumeBo recordBo = BeanUtil.copyProperties(consumptionBo, RemoteConsumeBo.class);
+
+        // String time = data.get("consumeDate").toString();
+        // recordBo.setConsumeMoney(new BigDecimal(data.get("consumeValue").toString()));
+        // recordBo.setConsumeDate(DateUtil.parse(time));
+        // recordBo.setCardNo(Long.valueOf(data.get("cardNo").toString()));
+        // recordBo.setFactoryId(Long.valueOf(data.get("factoryId").toString()));
+        // recordBo.setTermNo(Long.valueOf( data.get("termNo").toString()));
+        // recordBo.setTermRecordId(Long.valueOf(data.get("termRecordID").toString()));
+        // recordBo.setRealName(data.get("realName").toString());
+        // recordBo.setUserNumb(data.get("userNumb").toString());
+        // recordBo.setRecordStatus(Long.valueOf(data.get("posRecordState").toString()));
+        // recordBo.setCreditType(CreditTypeEnum.TERM_CONSUME);
+        // recordBo.setTenantId(DefaultConstants.TENANT_ID);
+        // recordBo.setUseType(SystemUseTypeEnum.CONSUME);
+
+        remoteConsumeService.dealKafkaConsumeData(recordBo);
+    }
+}