feature: 酒店系统

1.报到成功后写kafka修改

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

@@ -28,11 +28,9 @@ import org.dromara.backstage.payment.domain.vo.PtBagVo;
 import org.dromara.backstage.payment.service.IPtBagService;
 import org.dromara.backstage.payment.service.IPtUserAccountService;
 import org.dromara.common.core.constant.CacheNames;
-import org.dromara.common.core.constant.DefaultConstants;
 import org.dromara.common.core.constant.HotelBusinessConstants;
 import org.dromara.common.core.domain.R;
 import org.dromara.common.core.domain.model.ErrorInfo;
-import org.dromara.common.core.enums.HotelRoomStatusEnum;
 import org.dromara.common.core.enums.ResultCodeEnum;
 import org.dromara.common.core.enums.RoomCardTypeEnum;
 import org.dromara.common.message.kafka.domain.KafkaHeader;
@@ -184,7 +182,7 @@ public class TraineeBusiness {
         BeanUtil.copyProperties(ycTraineeVo, traineeVo);
 
         // 发送报到的kafka消息
-        sendCheckInMessageToKafka(bo, ycTraineeVo);
+        sendCheckInMessageToKafka(bo);
         return R.ok();
     }
 
@@ -355,10 +353,10 @@ public class TraineeBusiness {
 
     /**
      * 发送报到信息到kafka
+     *
      * @param bo 学员报到业务对象
-     * @param ycTraineeVo 学员信息
      */
-    public void sendCheckInMessageToKafka(YcTraineeBo bo, YcTraineeVo ycTraineeVo){
+    public void sendCheckInMessageToKafka(YcTraineeBo bo){
         TraineeCheckIn traineeCheckIn = new TraineeCheckIn();
         traineeCheckIn.setOtherSysId(bo.getOtherUserId());
         traineeCheckIn.setClassId(bo.getOtherDeptId());
@@ -376,6 +374,6 @@ public class TraineeBusiness {
         message.setHeader(header);
         message.setBody(traineeCheckIn);
 
-        kafkaProducer.sendKafkaMessage("eventBus",bo.getGuestId(),message);
+        kafkaProducer.sendKafkaMessage("eventBus",bo.getGuestId(),message,false);
     }
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/controller/self/TraineeController.java

@@ -116,7 +116,7 @@ public class TraineeController {
 
     @PostMapping("/api/v1/send/kafka")
     public ReturnResult sendKafka(@RequestBody YcTraineeBo bo) {
-        traineeBusiness.sendCheckInMessageToKafka(bo,null);
+        traineeBusiness.sendCheckInMessageToKafka(bo);
 
         return ReturnResult.success();
     }

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

@@ -29,20 +29,49 @@ public class KafkaNormalProducer {
     /**
      * Send.
      *
-     * @param topic   the topic
-     * @param message the message
+     * @param topic the topic
+     * @param data  the message
      */
-    public void send(String topic , String message) {
-        log.debug("发送消息到kafka消息系统， message:" + message);
-        kafkaTemplate.send(topic, message);
-        log.debug("发送消息到kafka消息系统结束");
+    public void sendKafkaMessage(String topic, Long createBy, KafkaMessage<?> data, Boolean save) {
+        String jsonMessage = JSON.toJSONString(data);
+        try {
+            ProducerRecord<String, String> record;
+            if (save) {
+                record = new ProducerRecord<>(topic, "YKT-SYNC-Message", jsonMessage);
+            } else {
+                record = new ProducerRecord<>(topic, jsonMessage);
+            }
+            log.info("发送同步数据到kafka消息系统, data: {}", jsonMessage);
+            CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(record);
+            send.whenComplete((result, ex) -> {
+                if (ex != null) {
+                    log.error("同步数据发送到kafka消息系统异常，data: {}", jsonMessage, ex);
+                    if (save) {
+                        // 异常信息入库
+                        insertRecord("F", createBy, data);
+                    }
+                } else {
+                    log.info("同步数据发送到kafka消息系统成功，data: {}", jsonMessage);
+                    if (save) {
+                        // 异常信息入库
+                        insertRecord("S", createBy, data);
+                    }
+                }
+            });
+        } catch (Exception e) {
+            log.error("同步数据发送到kafka消息系统异常，data: {}", data, e);
+            if (save) {
+                // 异常信息入库
+                insertRecord("S", createBy, data);
+            }
+        }
     }
 
 
-
-    public void sendKafkaMessage(String topic,Long createBy, KafkaMessage<?> data){
-        try{
-            String jsonMessage = JSON.toJSONString(data);
+    public void sendKafkaMessage(String topic, Long createBy, KafkaMessage<?> data) {
+        this.sendKafkaMessage(topic, createBy, data, true);
+        //try {
+        //String jsonMessage = JSON.toJSONString(data);
             /*KafkaMessage kafkaMessage = JsonUtils.parseObject(jsonMessage, KafkaMessage.class);
             Object body = kafkaMessage.getBody();
             if(body instanceof String){
@@ -58,30 +87,30 @@ public class KafkaNormalProducer {
                 System.err.println("body is other");
 
             }*/
-            ProducerRecord<String, String> record = new ProducerRecord<>(topic, "YKT-SYNC-Message", jsonMessage);
-            log.info("发送同步数据到kafka消息系统, data: " + jsonMessage);
-            CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(record);
-            send.whenComplete((result, ex) -> {
-                if (ex != null) {
-                    log.error("同步数据发送到kafka消息系统异常，data: " + jsonMessage, ex);
-
-                    // 异常信息入库
-                    insertRecord("F",createBy, data);
-                } else {
-                    log.info("同步数据发送到kafka消息系统成功，data: " + jsonMessage);
-                    insertRecord("S",createBy, data);
-                }
-            });
-        }catch (Exception e){
-            log.error("同步数据发送到kafka消息系统异常，data: " + data, e);
-            insertRecord("F",createBy, data);
-        }
+        //    ProducerRecord<String, String> record = new ProducerRecord<>(topic, "YKT-SYNC-Message", jsonMessage);
+        //    log.info("发送同步数据到kafka消息系统, data: " + jsonMessage);
+        //    CompletableFuture<SendResult<String, String>> send = kafkaTemplate.send(record);
+        //    send.whenComplete((result, ex) -> {
+        //        if (ex != null) {
+        //            log.error("同步数据发送到kafka消息系统异常，data: " + jsonMessage, ex);
+        //
+        //            // 异常信息入库
+        //            insertRecord("F", createBy, data);
+        //        } else {
+        //            log.info("同步数据发送到kafka消息系统成功，data: " + jsonMessage);
+        //            insertRecord("S", createBy, data);
+        //        }
+        //    });
+        //} catch (Exception e) {
+        //    log.error("同步数据发送到kafka消息系统异常，data: {}", data, e);
+        //    insertRecord("F", createBy, data);
+        //}
 
     }
 
     //记录入库
-    public void insertRecord(String result,Long createBy, KafkaMessage<?> data){
-        try{
+    public void insertRecord(String result, Long createBy, KafkaMessage<?> data) {
+        try {
             KafkaHeader header = data.getHeader();
             String eventId = header.getEventId();
             String sender = header.getSender();
@@ -95,8 +124,8 @@ public class KafkaNormalProducer {
             bo.setTenantId(tenantId);
             bo.setResult(result);
             String s = JSON.toJSONString(data);
-            if(StringUtils.isNotBlank(s) && s.length()> 5000){
-                s = s.substring(0,5000);
+            if (StringUtils.isNotBlank(s) && s.length() > 5000) {
+                s = s.substring(0, 5000);
             }
             bo.setMessage(s);
 //            LoginUser loginUser = LoginHelper.getLoginUser();
@@ -106,7 +135,7 @@ public class KafkaNormalProducer {
             bo.setTenantId(tenantId);
             bo.setCreateTime(new Date());
             sendMessageRecordService.insertByBo(bo);
-        }catch (Exception e){
+        } catch (Exception e) {
             log.error("kafka消息记录入库异常，data: " + data, e);
         }
     }