卡务中心双向同步

ruoyi-common/ruoyi-common-core/src/main/java/org/dromara/common/core/constant/CloudKafkaEventConstants.java

@@ -0,0 +1,23 @@
+package org.dromara.common.core.constant;
+
+/**
+ * name: CloudKafkaEventConstants
+ * package: org.dromara.common.core.constant
+ * description: 云端消息推送事件常量
+ * date: 2025-01-02 15:05:08 15:05
+ *
+ * @author luoyibo
+ * @version 0.1
+ * @since JDK 1.8
+ */
+public class CloudKafkaEventConstants {
+    // 业务系统编号
+    public static final String SENDER = "000";
+    // 消费记录
+    public static final String CONSUME_DETAIL = SENDER + "00001";
+    //部门信息
+    public static final String DEPT_INFO = SENDER + "00002";
+    //人员信息
+    public static final String USER_INFO = SENDER + "00003";
+
+}

ruoyi-common/ruoyi-common-message/src/main/java/org/dromara/common/message/kafka/aop/aspect/SyncDataToLocalAspect.java

@@ -10,6 +10,7 @@ import org.aspectj.lang.JoinPoint;
 import org.aspectj.lang.annotation.AfterReturning;
 import org.aspectj.lang.annotation.Aspect;
 import org.aspectj.lang.reflect.CodeSignature;
+import org.dromara.common.core.api.ReturnResult;
 import org.dromara.common.message.kafka.aop.annotation.SyncDataToLocal;
 import org.dromara.backstage.api.RemoteSyncToLocalByKafkaService;
 import org.dromara.common.core.domain.R;
@@ -54,7 +55,13 @@ public class SyncDataToLocalAspect {
             }else{
                 log.error("同步数据消息未发送：controller 方法返回结果未失败！");
             }
-        }else{
+        }else if (jsonResult instanceof ReturnResult r){
+            if (r.isSuccess()) {
+                sendSyncMessage(joinPoint, syncDataToLocal);
+            }else{
+                log.error("同步数据消息未发送：controller 方法返回结果未失败！");
+            }
+        } else {
             log.error("同步数据消息未发送：controller 方法返回不是R类型！");
         }
 
@@ -107,9 +114,12 @@ public class SyncDataToLocalAspect {
             }else{
                 data.setBody(null);
             }
-            LoginUser loginUser = LoginHelper.getLoginUser();
-
-            kafkaProducer.sendAndInsert(KafkaTopicConstants.SYNC_DATA_TOPIC, loginUser.getUserId(),data);
+            try {
+                LoginUser loginUser = LoginHelper.getLoginUser();
+                kafkaProducer.sendAndInsert(KafkaTopicConstants.SYNC_DATA_TOPIC, loginUser.getUserId(),data);
+            } catch (Exception e){
+                kafkaProducer.sendAndInsert(KafkaTopicConstants.SYNC_DATA_TOPIC, 100L,data);
+            }
         }catch (Exception e){
             log.error("同步数据消息未发送：发送消息异常！", e);
         }

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

@@ -0,0 +1,38 @@
+package org.dromara.system.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);
+        }
+    }
+}