一卡通数据双向同步本地消费luoji

ruoyi-server/ruoyi-server-base/src/main/java/org/dromara/server/base/service/yktOperation/SyncRemotePtParameterService.java

@@ -0,0 +1,18 @@
+package org.dromara.server.base.service.yktOperation;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * 系统参数同步
+ */
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class SyncRemotePtParameterService {
+
+    public void addPtParameter(){
+
+    }
+}

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

@@ -8,3 +8,4 @@ org.dromara.server.base.service.user.strategy.SyncUserStrategyContent
 org.dromara.server.base.service.user.strategy.impl.SyncTeacherStrategyImpl
 org.dromara.server.base.service.user.strategy.impl.SyncGraduateStrategyImpl
 org.dromara.server.base.service.user.strategy.impl.SyncTraineeStrategyImpl
+org.dromara.server.base.service.yktOperation.SyncRemotePtParameterService

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/constant/kafka/YktOperationEventConstraints.java

@@ -0,0 +1,26 @@
+package org.dromara.server.mq.constant.kafka;
+
+
+public class YktOperationEventConstraints {
+
+    /**
+     * 系统参数功能模块标识
+     */
+    public static final String PARAMETER_SENDER = "YKT_101";
+
+    /**
+     * 系统参数新增
+     */
+    public static final String PARAMETER_ADD = PARAMETER_SENDER + "_ADD";
+
+    /**
+     * 系统参数修改
+     */
+    public static final String PARAMETER_EDIT = PARAMETER_SENDER + "_EDIT";
+
+    /**
+     * 系统参数删除
+     */
+    public static final String PARAMETER_DEL = PARAMETER_SENDER + "_DEL";
+
+}

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

@@ -8,6 +8,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.dromara.common.message.kafka.domain.KafkaMessage;
 import org.dromara.server.mq.event.kafka.EventStrategyContext;
+import org.dromara.server.mq.event.kafka.YktEventStrategyContext;
 import org.springframework.kafka.annotation.KafkaListener;
 
 /**
@@ -25,6 +26,7 @@ import org.springframework.kafka.annotation.KafkaListener;
 //@Component
 public class KafkaConsumer {
     private final EventStrategyContext eventStrategyContext;
+    private final YktEventStrategyContext yktEventStrategyContext;
     @KafkaListener(topics = "eventBus", groupId = "test-group-id")
     public void kafkaReceiveHandler(ConsumerRecord<String, String> record) {
         KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
@@ -36,6 +38,26 @@ public class KafkaConsumer {
             eventStrategyContext.doMsgHandle(eventType, eventMsg);
         } catch (Exception e) {
             log.error("[kafka消息处理失败]-[消息:{}-[错误:{}]", receiveMsg, e.getMessage());
+        }
+    }
+
+    /**
+     * 一卡通云端业务操作本地同步处理
+     * @param record kafka消息
+     */
+    @KafkaListener(topics = "ykt_operation", groupId = "ykt_local_listener")
+    public void cloudOperationSync(ConsumerRecord<String, String> record){
+        KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
+        log.info("[接收到Kafka消息]-[{}]", receiveMsg);
+        try{
+            String eventType = receiveMsg.getHeader().getEventType();
+    //          String tenantId = receiveMsg.getHeader().getTenantId();
+            String tenantId = "";
+            JSONObject eventMsg = JSONUtil.parseObj(receiveMsg.getBody());
+            yktEventStrategyContext.doMsgHandle(eventType, eventMsg);
+
+        } catch (Exception e){
+            log.error("[kafka消息处理失败]-[消息:{}-[错误:{}]", receiveMsg, e.getMessage());
 
         }
     }

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/event/kafka/IYktEventStrategy.java

@@ -0,0 +1,15 @@
+package org.dromara.server.mq.event.kafka;
+
+import cn.hutool.json.JSONObject;
+
+/**
+ * 一卡通统一事件处理接口
+ */
+public interface IYktEventStrategy {
+    /**
+     *
+     * @param eventType 事件类型
+     * @param msg 消息体
+     */
+    void doMsgHandle(String eventType,JSONObject msg);
+}

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/event/kafka/YktEventStrategyContext.java

@@ -0,0 +1,25 @@
+package org.dromara.server.mq.event.kafka;
+
+import cn.hutool.json.JSONObject;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 一卡通内部操作事件处理上下文
+ */
+@Service
+public class YktEventStrategyContext {
+    private final Map<String, IYktEventStrategy> strategyMap = new ConcurrentHashMap<>();
+
+    @Autowired
+    public YktEventStrategyContext(Map<String, IYktEventStrategy> strategyMap) {
+        this.strategyMap.putAll(strategyMap);
+    }
+
+    public void doMsgHandle(String eventType, JSONObject msg) {
+        strategyMap.get(eventType).doMsgHandle(eventType, msg);
+    }
+}

ruoyi-server/ruoyi-server-mqdata/src/main/java/org/dromara/server/mq/event/kafka/impl/yktOperation/PtParameterEventStrategyImpl.java

@@ -0,0 +1,44 @@
+package org.dromara.server.mq.event.kafka.impl.yktOperation;
+
+import cn.hutool.json.JSONObject;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.server.base.service.dept.SyncRemoteDeptService;
+import org.dromara.server.base.service.yktOperation.SyncRemotePtParameterService;
+import org.dromara.server.mq.constant.kafka.YktOperationEventConstraints;
+import org.dromara.server.mq.event.kafka.IYktEventStrategy;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author Hz
+ * @date 2024/10/29
+ * @description 系统参数功能同步策略
+ */
+@Slf4j
+@RequiredArgsConstructor
+@Service(YktOperationEventConstraints.PARAMETER_SENDER)
+public class PtParameterEventStrategyImpl implements IYktEventStrategy {
+
+    private final SyncRemotePtParameterService parameterService;
+    @Override
+    public void doMsgHandle(String eventType, JSONObject msg) {
+        switch (eventType) {
+            case YktOperationEventConstraints.PARAMETER_ADD:{
+                log.info("新增系统参数");
+            }
+            break;
+            case YktOperationEventConstraints.PARAMETER_EDIT:{
+                log.info("修改系统参数");
+            }
+            break;
+            case YktOperationEventConstraints.PARAMETER_DEL:{
+                log.info("删除系统参数");
+            }
+            break;
+            default:
+                log.info("未知事件");
+
+        }
+    }
+
+}