数据同步消息发送注解与逻辑

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/aop/annotation/SyncDataToLocal.java

@@ -7,6 +7,7 @@ import java.lang.annotation.*;
  *
  * @author bing
  */
+@Inherited
 @Target({ElementType.METHOD})
 @Retention(RetentionPolicy.RUNTIME)
 @Documented

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/aop/aspect/SyncDataToLocalAspect.java

@@ -15,7 +15,9 @@ 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.common.satoken.utils.LoginHelper;
+import org.dromara.system.api.model.LoginUser;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.stereotype.Component;
 
 import java.util.HashMap;
 import java.util.Map;
@@ -28,7 +30,7 @@ import java.util.Map;
 @RequiredArgsConstructor
 @Slf4j
 @Aspect
-@AutoConfiguration
+@Component
 public class SyncDataToLocalAspect {
 
     private final KafkaProducer kafkaProducer;
@@ -39,11 +41,12 @@ public class SyncDataToLocalAspect {
      *
      * @param joinPoint 切点
      */
-    @AfterReturning(pointcut = "@annotation(controllerSyncData2Local)", returning = "jsonResult")
-    public void doAfterReturning(JoinPoint joinPoint, SyncDataToLocal controllerSyncData2Local, Object jsonResult) {
+    @AfterReturning(pointcut = "@annotation(syncDataToLocal)", returning = "jsonResult")
+    public void doAfterReturning(JoinPoint joinPoint, SyncDataToLocal syncDataToLocal, Object jsonResult) {
+//        System.err.println("进入呢么？");
         if (jsonResult instanceof R<?> r) {
             if (r.getCode() == R.SUCCESS) {
-                sendSyncMessage(joinPoint, controllerSyncData2Local);
+                sendSyncMessage(joinPoint, syncDataToLocal);
             }else{
                 log.error("同步数据消息未发送：controller 方法返回结果未失败！");
             }
@@ -86,7 +89,8 @@ public class SyncDataToLocalAspect {
             }else{
                 data.setBody(null);
             }
-            kafkaProducer.sendKafkaMessage(KafkaTopicConstants.SYNC_DATA_TOPIC, data);
+            LoginUser loginUser = LoginHelper.getLoginUser();
+            kafkaProducer.sendKafkaMessage(KafkaTopicConstants.SYNC_DATA_TOPIC, loginUser.getUserId(),data);
         }catch (Exception e){
             log.error("同步数据消息未发送：发送消息异常！", e);
         }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/controller/PtParameterController.java

@@ -6,6 +6,8 @@ import lombok.RequiredArgsConstructor;
 import jakarta.servlet.http.HttpServletResponse;
 import jakarta.validation.constraints.*;
 import cn.dev33.satoken.annotation.SaCheckPermission;
+import org.dromara.backstage.aop.annotation.SyncDataToLocal;
+import org.dromara.common.message.kafka.constant.MessageEventTypeConstants;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.validation.annotation.Validated;
 import org.dromara.common.idempotent.annotation.RepeatSubmit;
@@ -75,22 +77,12 @@ public class PtParameterController extends BaseController {
     @SaCheckPermission("basicParameter:ptParameter:add")
     @Log(title = "系统参数", businessType = BusinessType.INSERT)
     @RepeatSubmit()
+    @SyncDataToLocal(eventType = MessageEventTypeConstants.ptParameter_ADD)
     @PostMapping()
     public R<Void> add(@Validated(AddGroup.class) @RequestBody PtParameterBo bo) {
         return toAjax(ptParameterService.insertByBo(bo));
     }
 
-    /**
-     * 修改系统参数
-     */
-    @SaCheckPermission("basicParameter:ptParameter:edit")
-    @Log(title = "系统参数", businessType = BusinessType.UPDATE)
-    @RepeatSubmit()
-    @PutMapping()
-    public R<Void> edit(@Validated(EditGroup.class) @RequestBody PtParameterBo bo) {
-        return toAjax(ptParameterService.updateByBo(bo));
-    }
-
     /**
      * 删除系统参数
      *
@@ -99,10 +91,23 @@ public class PtParameterController extends BaseController {
     @SaCheckPermission("basicParameter:ptParameter:remove")
     @Log(title = "系统参数", businessType = BusinessType.DELETE)
     @DeleteMapping("/{paramIds}")
+    @SyncDataToLocal(eventType = MessageEventTypeConstants.ptParameter_EDIT)
     public R<Void> remove(@NotEmpty(message = "主键不能为空")
                           @PathVariable Long[] paramIds) {
         return toAjax(ptParameterService.deleteWithValidByIds(List.of(paramIds), true));
     }
+
+    /**
+     * 修改系统参数
+     */
+    @SaCheckPermission("basicParameter:ptParameter:edit")
+    @Log(title = "系统参数", businessType = BusinessType.UPDATE)
+    @RepeatSubmit()
+    @SyncDataToLocal(eventType = MessageEventTypeConstants.ptParameter_REMOVE)
+    @PutMapping()
+    public R<Void> edit(@Validated(EditGroup.class) @RequestBody PtParameterBo bo) {
+        return toAjax(ptParameterService.updateByBo(bo));
+    }
     /**
      * 根据参数键名查询参数值
      *

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/domain/SendMessageRecord.java

@@ -6,6 +6,8 @@ import lombok.Data;
 import lombok.EqualsAndHashCode;
 
 import java.io.Serial;
+import java.io.Serializable;
+import java.util.Date;
 
 /**
  * 消息发送记录对象 t_send_message_record
@@ -14,9 +16,8 @@ import java.io.Serial;
  * @date 2024-10-30
  */
 @Data
-@EqualsAndHashCode(callSuper = true)
 @TableName("t_send_message_record")
-public class SendMessageRecord extends TenantEntity {
+public class SendMessageRecord implements Serializable {
 
     @Serial
     private static final long serialVersionUID = 1L;
@@ -61,5 +62,14 @@ public class SendMessageRecord extends TenantEntity {
      */
     private String sender;
 
+    /**
+     * 租户编号
+     */
+    private String tenantId;
+
+    private Long createBy;
+
+    private Date createTime;
+
 
 }

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

@@ -43,7 +43,7 @@ public class KafkaProducer {
 
 
 
-    public void sendKafkaMessage(String topic, KafkaMessage<?> data){
+    public void sendKafkaMessage(String topic,Long createBy, KafkaMessage<?> data){
         try{
             String jsonMessage = JSON.toJSONString(data);
             ProducerRecord<String, String> record = new ProducerRecord<>(topic, "YKT-SYNC-Message", jsonMessage);
@@ -54,21 +54,21 @@ public class KafkaProducer {
                     log.error("同步数据发送到kafka消息系统异常，data: " + jsonMessage, ex);
 
                     // 异常信息入库
-                    insertRecord("F", data);
+                    insertRecord("F",createBy, data);
                 } else {
                     log.info("同步数据发送到kafka消息系统成功，data: " + jsonMessage);
-                    insertRecord("S", data);
+                    insertRecord("S",createBy, data);
                 }
             });
         }catch (Exception e){
             log.error("同步数据发送到kafka消息系统异常，data: " + data, e);
-            insertRecord("F", data);
+            insertRecord("F",createBy, data);
         }
 
     }
 
     //记录入库
-    public void insertRecord(String result, KafkaMessage<?> data){
+    public void insertRecord(String result,Long createBy, KafkaMessage<?> data){
         try{
             KafkaHeader header = data.getHeader();
             String eventId = header.getEventId();
@@ -83,8 +83,8 @@ public class KafkaProducer {
             bo.setTenantId(tenantId);
             bo.setResult(result);
             bo.setMessage(JSON.toJSONString(data));
-            LoginUser loginUser = LoginHelper.getLoginUser();
-            bo.setCreateBy(loginUser.getUserId());
+//            LoginUser loginUser = LoginHelper.getLoginUser();
+            bo.setCreateBy(createBy);
             bo.setMqType("kafka");
             bo.setTopic(KafkaTopicConstants.SYNC_DATA_TOPIC);
             bo.setTenantId(tenantId);