补助设置数据双向同步功能开发

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/RemotePtSubsidyService.java

@@ -0,0 +1,23 @@
+package org.dromara.backstage.api;
+
+import org.dromara.backstage.api.domain.bo.RemotePtSubsidyBo;
+
+import java.util.Collection;
+
+/**
+ * description: 补助设置服务
+ *
+ * @author luoyibo
+ * @version 0.1
+ * @since JDK 1.8
+ */
+public interface RemotePtSubsidyService {
+
+    Boolean insertByBo(RemotePtSubsidyBo bo) throws Exception;
+
+    Boolean updateByBo(RemotePtSubsidyBo bo) throws Exception;
+
+    Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid) throws Exception;
+
+    Boolean deleteSubItemByIds(Collection<Long> ids);
+}

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/RemoteSendMessageRecordService.java

@@ -0,0 +1,15 @@
+package org.dromara.backstage.api;
+
+import org.dromara.backstage.api.domain.bo.RemoteSendMessageRecordBo;
+
+/**
+ * 消息发送记录接口
+ *
+ * @author Lion Li
+ */
+public interface RemoteSendMessageRecordService {
+
+    Boolean updateByBo(RemoteSendMessageRecordBo bo) throws Exception;
+
+
+}

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/domain/bo/RemotePtSubsidyBo.java

@@ -0,0 +1,79 @@
+package org.dromara.backstage.api.domain.bo;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.Date;
+import java.util.List;
+
+/**
+ * 补助设置业务对象 t_pt_subsidy
+ *
+ * @author Yz
+ * @date 2024-08-12
+ */
+@Data
+@NoArgsConstructor
+public class RemotePtSubsidyBo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1844034081728852725L;
+    /**
+     * ID
+     */
+    private Long mainId;
+
+    /**
+     * 补助名称
+     */
+    private String subsidyName;
+
+    /**
+     * 补助类型
+     */
+    private String subsidyType;
+
+    /**
+     * 补助金额
+     */
+    private Long fillMoney;
+
+    /**
+     * 预计到账时间
+     */
+    private Date fillDate;
+
+    /**
+     * 入账钱包
+     */
+    private Long bagCode;
+
+    /**
+     * 到账状态
+     */
+    private String fillStatus;
+
+    /**
+     * 审核状态
+     */
+    private String auditStatus;
+
+    /**
+     * 审核人
+     */
+    private Long auditUser;
+
+    /**
+     * 审核时间
+     */
+    private Date auditDate;
+
+    /**
+     * 备注
+     */
+    private String remark;
+
+    private List<RemotePtSubsidyitemBo> ptSubsidyItemList;
+}

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/domain/bo/RemotePtSubsidyitemBo.java

@@ -0,0 +1,99 @@
+package org.dromara.backstage.api.domain.bo;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serial;
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 补助明细业务对象
+ *
+ * @author Yz
+ * @date 2024-08-14
+ */
+@Data
+@NoArgsConstructor
+public class RemotePtSubsidyitemBo implements Serializable {
+
+    @Serial
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 明细Id,主键
+     */
+    private Long itemId;
+
+    /**
+     * 补助设置Id
+     */
+    private Long mainId;
+
+    /**
+     * 补助人员Id
+     */
+    private Long userId;
+
+    /**
+     * 补助类型
+     */
+    private String subsidyType;
+
+    /**
+     * 补助金额
+     */
+    private Long fillMoney;
+
+    /**
+     * 预计到账时间
+     */
+    private Date fillDate;
+
+    /**
+     * 领取状态
+     */
+    private String fillStatus;
+
+    /**
+     * 领取时间
+     */
+    private Date getDate;
+
+    /**
+     * 领取补助之后钱包余额
+     */
+    private Long beforeMoney;
+
+    /**
+     * 领取补助之后钱包余额
+     */
+    private Long afterMoney;
+
+    /**
+     * 原始记录Id,一般是指写入的充值备份表主键
+     */
+    private String originalId;
+
+    /**
+     * 用户编号/工号
+     */
+    private String userNumb;
+
+    /**
+     * 用户姓名
+     */
+    private String realName;
+
+    /**
+     * 部门名称
+     */
+    private String deptName;
+
+    /**
+     * 备注
+     */
+    private String remark;
+
+
+}

ruoyi-api/ruoyi-api-backstage/src/main/java/org/dromara/backstage/api/domain/bo/RemoteSendMessageRecordBo.java

@@ -0,0 +1,63 @@
+package org.dromara.backstage.api.domain.bo;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * 消息发送记录业务对象
+ *
+ * @author bing
+ * @date 2024-10-30
+ */
+@Data
+@NoArgsConstructor
+public class RemoteSendMessageRecordBo implements Serializable {
+
+    /**
+     * 主键id
+     */
+    private Long recordId;
+
+    /**
+     * 消息类型：kafka、rabbitmq、rocketmq
+     */
+    private String mqType;
+
+    /**
+     * 消息主题
+     */
+    private String topic;
+
+    /**
+     * 消息事件类型
+     */
+    private String eventType;
+
+    /**
+     * 发送结果：S 成功，F 失败
+     */
+    private String result;
+
+    /**
+     * 消息
+     */
+    private String message;
+
+    /**
+     * $column.columnComment
+     */
+    private String eventId;
+
+    /**
+     * 发送方
+     */
+    private String sender;
+
+    /**
+     * 消费状态
+     */
+    private String consumeStatus;
+}

ruoyi-common/ruoyi-common-message/src/main/java/org/dromara/common/message/kafka/constant/MessageEventTypeConstants.java

@@ -138,12 +138,10 @@ public class MessageEventTypeConstants {
     /** 补助设置新增 */
     public static final String	SUBSIDY_ADD = "YKT_102_ADD";
 
-    /** 补助设置修改 */
+    /** 补助设置修改、审核 */
     public static final String	SUBSIDY_EDIT = "YKT_102_EDIT";
     /** 补助设置删除 */
     public static final String	SUBSIDY_REMOVE = "YKT_102_REMOVE";
-    /** 补助设置审核 和编辑是同一个接口*/
-    public static final String	SUBSIDY_AUDIT = "YKT_102_AUDIT";
     /** 补助明细人员删除 */
     public static final String	SUBSIDY_ITEM_REMOVE = "YKT_102_ITEM_REMOVE";
     /** 消费设备新增 */

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

@@ -1,9 +1,7 @@
 package org.dromara.backstage.basics.domain;
 
-import org.dromara.common.tenant.core.TenantEntity;
-import com.baomidou.mybatisplus.annotation.*;
+import com.baomidou.mybatisplus.annotation.TableName;
 import lombok.Data;
-import lombok.EqualsAndHashCode;
 
 import java.io.Serial;
 import java.io.Serializable;
@@ -71,5 +69,7 @@ public class SendMessageRecord implements Serializable {
 
     private Date createTime;
 
+    private String consumeStatus;
+    private Date updateTime;
 
 }

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

@@ -69,5 +69,9 @@ public class SendMessageRecordBo extends TenantEntity {
     @NotBlank(message = "发送方不能为空", groups = { AddGroup.class, EditGroup.class })
     private String sender;
 
+    /**
+     * 消费状态
+     */
+    private String consumeStatus;
 
 }

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

@@ -77,5 +77,9 @@ public class SendMessageRecordVo implements Serializable {
     @ExcelProperty(value = "发送方")
     private String sender;
 
-
+    /**
+     * 消费状态
+     */
+    @ExcelProperty(value = "消费状态")
+    private String consumeStatus;
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/dubbo/RemotePtSubsidyServiceImpl.java

@@ -0,0 +1,47 @@
+package org.dromara.backstage.basics.dubbo;
+
+import cn.hutool.core.bean.BeanUtil;
+import lombok.RequiredArgsConstructor;
+import org.apache.dubbo.config.annotation.DubboService;
+import org.dromara.backstage.api.RemotePtSubsidyService;
+import org.dromara.backstage.api.domain.bo.RemotePtSubsidyBo;
+import org.dromara.backstage.cardCenter.domain.bo.PtSubsidyBo;
+import org.dromara.backstage.cardCenter.service.IPtSubsidyService;
+import org.dromara.backstage.cardCenter.service.IPtSubsidyitemService;
+import org.springframework.stereotype.Service;
+
+import java.util.Collection;
+
+/**
+ * 补助设置服务实现类
+ *
+ * @author Lion Li
+ */
+@RequiredArgsConstructor
+@Service
+@DubboService
+public class RemotePtSubsidyServiceImpl implements RemotePtSubsidyService {
+
+    private final IPtSubsidyService subsidyService;
+    private final IPtSubsidyitemService subsidyItemService;
+
+    @Override
+    public Boolean insertByBo(RemotePtSubsidyBo bo) throws Exception {
+        return subsidyService.insertByBo(BeanUtil.copyProperties(bo, PtSubsidyBo.class));
+    }
+
+    @Override
+    public Boolean updateByBo(RemotePtSubsidyBo bo) throws Exception {
+        return subsidyService.updateByBo(BeanUtil.copyProperties(bo, PtSubsidyBo.class));
+    }
+
+    @Override
+    public Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid) throws Exception {
+        return subsidyService.deleteWithValidByIds(ids,isValid);
+    }
+
+    @Override
+    public Boolean deleteSubItemByIds(Collection<Long> ids) {
+        return subsidyItemService.deleteWithValidByIds(ids,false);
+    }
+}

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/dubbo/RemoteSendMessageRecordServiceImpl.java

@@ -0,0 +1,24 @@
+package org.dromara.backstage.basics.dubbo;
+
+
+import cn.hutool.core.bean.BeanUtil;
+import lombok.RequiredArgsConstructor;
+import org.apache.dubbo.config.annotation.DubboService;
+import org.dromara.backstage.api.RemoteSendMessageRecordService;
+import org.dromara.backstage.api.domain.bo.RemoteSendMessageRecordBo;
+import org.dromara.backstage.basics.domain.bo.SendMessageRecordBo;
+import org.dromara.backstage.basics.service.ISendMessageRecordService;
+import org.springframework.stereotype.Service;
+
+@RequiredArgsConstructor
+@Service
+@DubboService
+public class RemoteSendMessageRecordServiceImpl implements RemoteSendMessageRecordService {
+
+    private final ISendMessageRecordService sendMessageRecordService;
+
+    @Override
+    public Boolean updateByBo(RemoteSendMessageRecordBo bo) throws Exception {
+        return sendMessageRecordService.updateBoByEventId(BeanUtil.copyProperties(bo, SendMessageRecordBo.class));
+    }
+}

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/mapper/SendMessageRecordMapper.java

@@ -11,5 +11,4 @@ import org.dromara.common.mybatis.core.mapper.BaseMapperPlus;
  * @date 2024-10-30
  */
 public interface SendMessageRecordMapper extends BaseMapperPlus<SendMessageRecord, SendMessageRecordVo> {
-
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/service/ISendMessageRecordService.java

@@ -66,4 +66,9 @@ public interface ISendMessageRecordService {
      * @return 是否删除成功
      */
     Boolean deleteWithValidByIds(Collection<Long> ids, Boolean isValid);
+
+    /**
+     * 根据事件ID修改消息发送记录
+     */
+    Boolean updateBoByEventId(SendMessageRecordBo bo);
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/basics/service/impl/SendMessageRecordServiceImpl.java

@@ -1,5 +1,6 @@
 package org.dromara.backstage.basics.service.impl;
 
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
 import org.dromara.common.core.utils.MapstructUtils;
 import org.dromara.common.core.utils.StringUtils;
 import org.dromara.common.mybatis.core.page.TableDataInfo;
@@ -16,6 +17,7 @@ import org.dromara.backstage.basics.domain.SendMessageRecord;
 import org.dromara.backstage.basics.mapper.SendMessageRecordMapper;
 import org.dromara.backstage.basics.service.ISendMessageRecordService;
 
+import java.util.Date;
 import java.util.List;
 import java.util.Map;
 import java.util.Collection;
@@ -151,4 +153,13 @@ public class SendMessageRecordServiceImpl implements ISendMessageRecordService {
         }
         return baseMapper.deleteByIds(ids) > 0;
     }
+
+    @Override
+    public Boolean updateBoByEventId(SendMessageRecordBo bo) {
+        UpdateWrapper<SendMessageRecord> updateWrapper = new UpdateWrapper<>();
+        updateWrapper.eq("event_id",bo.getEventId());
+        SendMessageRecord update = MapstructUtils.convert(bo, SendMessageRecord.class);
+        update.setUpdateTime(new Date());
+        return baseMapper.update(update,updateWrapper) > 0;
+    }
 }

ruoyi-modules/ruoyi-backstage/src/main/java/org/dromara/backstage/cardCenter/service/impl/PtSubsidyServiceImpl.java

@@ -1,26 +1,28 @@
 package org.dromara.backstage.cardCenter.service.impl;
 
-import org.dromara.backstage.cardCenter.domain.PtSubsidyitem;
-import org.dromara.backstage.cardCenter.domain.bo.PtSubsidyitemBo;
-import org.dromara.common.core.utils.MapstructUtils;
-import org.dromara.common.core.utils.StringUtils;
-import org.dromara.common.mybatis.core.page.TableDataInfo;
-import org.dromara.common.mybatis.core.page.PageQuery;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
 import lombok.RequiredArgsConstructor;
-import org.springframework.stereotype.Service;
+import org.dromara.backstage.cardCenter.domain.PtSubsidy;
+import org.dromara.backstage.cardCenter.domain.PtSubsidyitem;
 import org.dromara.backstage.cardCenter.domain.bo.PtSubsidyBo;
+import org.dromara.backstage.cardCenter.domain.bo.PtSubsidyitemBo;
 import org.dromara.backstage.cardCenter.domain.vo.PtSubsidyVo;
-import org.dromara.backstage.cardCenter.domain.PtSubsidy;
 import org.dromara.backstage.cardCenter.mapper.PtSubsidyMapper;
+import org.dromara.backstage.cardCenter.mapper.PtSubsidyitemMapper;
 import org.dromara.backstage.cardCenter.service.IPtSubsidyService;
+import org.dromara.backstage.cardCenter.service.IPtSubsidyitemService;
+import org.dromara.common.core.utils.MapstructUtils;
+import org.dromara.common.core.utils.StringUtils;
+import org.dromara.common.mybatis.core.page.PageQuery;
+import org.dromara.common.mybatis.core.page.TableDataInfo;
+import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.Collection;
 import java.util.List;
 import java.util.Map;
-import java.util.Collection;
 
 /**
  * 补助设置Service业务层处理
@@ -33,7 +35,8 @@ import java.util.Collection;
 public class PtSubsidyServiceImpl implements IPtSubsidyService {
 
     private final PtSubsidyMapper baseMapper;
-    private final PtSubsidyitemServiceImpl subsidyitemService;
+    private final IPtSubsidyitemService subsidyitemService;
+    private final PtSubsidyitemMapper subsidyitemMapper;
     /**
      * 查询补助设置
      *
@@ -96,11 +99,14 @@ public class PtSubsidyServiceImpl implements IPtSubsidyService {
         boolean flag = baseMapper.insert(add) > 0;
         if (flag) {
             List<PtSubsidyitemBo> subsidyitemBoList = bo.getPtSubsidyItemList();
+
             bo.setMainId(add.getMainId());
             subsidyitemBoList.stream().forEach(subsidyitemBo ->{
-                subsidyitemBo.setMainId(add.getMainId());
-                subsidyitemService.insertByBo(subsidyitemBo);
+                setSubsidyInfo(subsidyitemBo, bo);
             });
+            Collection<PtSubsidyitem> list = MapstructUtils.convert(subsidyitemBoList, PtSubsidyitem.class);
+            subsidyitemMapper.insertBatch(list);
+
         }
         return flag;
     }
@@ -122,13 +128,9 @@ public class PtSubsidyServiceImpl implements IPtSubsidyService {
             List<PtSubsidyitemBo> subsidyitemBoList = bo.getPtSubsidyItemList();
             subsidyitemBoList.stream().forEach(subsidyitemBo ->{
                 setSubsidyInfo(subsidyitemBo, bo);
-                if(subsidyitemBo.getItemId() == null){
-                    subsidyitemBo.setMainId(bo.getMainId());
-                    subsidyitemService.insertByBo(subsidyitemBo);
-                }else {
-                    subsidyitemService.updateByBo(subsidyitemBo);
-                }
             });
+            Collection<PtSubsidyitem> list = MapstructUtils.convert(subsidyitemBoList, PtSubsidyitem.class);
+            subsidyitemMapper.insertOrUpdateBatch(list);
         }
         return flag;
     }

ruoyi-modules/ruoyi-backstage/src/main/resources/mapper/basics/SendMessageRecordMapper.xml

@@ -13,8 +13,10 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
             <result property="result"    column="result"    />
             <result property="createBy"    column="create_by"    />
             <result property="createTime"    column="create_time"    />
+            <result property="updateTime"    column="update_time"    />
             <result property="message"    column="message"    />
             <result property="eventId"    column="event_id"    />
             <result property="sender"    column="sender"    />
+        <result property="consumeStatus"    column="consume_status"    />
     </resultMap>
 </mapper>

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

@@ -0,0 +1,66 @@
+package org.dromara.server.base.service.yktOperation;
+
+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.RemotePtSubsidyService;
+import org.dromara.backstage.api.domain.bo.RemotePtSubsidyBo;
+import org.dromara.common.core.exception.ServiceException;
+import org.springframework.stereotype.Service;
+
+import java.util.Collection;
+
+/**
+ * 补助设置同步
+ */
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class SyncRemotePtSubsidyService {
+
+    @DubboReference
+    private final RemotePtSubsidyService subsidyService;
+
+    /**
+     * 远程调用补助设置新增
+     */
+    public void addPtSubsidy(Object msg) throws Exception{
+        boolean flag = subsidyService.insertByBo(JSONUtil.parseObj(msg).toBean(RemotePtSubsidyBo.class));
+        if (!flag){
+            throw new ServiceException("补助设置新增失败");
+        }
+    }
+
+    /**
+     * 远程调用补助设置修改
+     */
+    public void editPtSubsidy(Object msg) throws Exception{
+        boolean flag = subsidyService.updateByBo(JSONUtil.parseObj(msg).toBean(RemotePtSubsidyBo.class));
+        if (!flag){
+            throw new ServiceException("补助设置修改失败");
+        }
+    }
+
+    /**
+     * 远程调用补助设置删除
+     */
+    public void delPtSubsidy(Object msg) throws Exception{
+        Collection<Long> ids = JSONUtil.parseArray(msg).toBean(Collection.class);
+        boolean flag = subsidyService.deleteWithValidByIds(ids,false);
+        if (!flag){
+            throw new ServiceException("补助设置删除失败");
+        }
+    }
+
+    /**
+     * 远程调用补助明细删除
+     */
+    public void delPtSubsidyItem(Object msg) throws ServiceException{
+        Collection<Long> ids = JSONUtil.parseArray(msg).toBean(Collection.class);
+        boolean flag = subsidyService.deleteSubItemByIds(ids);
+        if (!flag){
+            throw new ServiceException("补助明细删除失败");
+        }
+    }
+}

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

@@ -0,0 +1,31 @@
+package org.dromara.server.base.service.yktOperation;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.dubbo.config.annotation.DubboReference;
+import org.dromara.backstage.api.RemoteSendMessageRecordService;
+import org.dromara.backstage.api.domain.bo.RemoteSendMessageRecordBo;
+import org.dromara.common.core.exception.ServiceException;
+import org.springframework.stereotype.Service;
+
+/**
+ * 更新消息记录
+ */
+@Service
+@Slf4j
+@RequiredArgsConstructor
+public class SyncRemoteSendMessageRecordService {
+
+    @DubboReference
+    private final RemoteSendMessageRecordService sendMessageRecordService;
+
+    /**
+     * 远程调用消息发送记录修改
+     */
+    public void editSendMessageRecord(RemoteSendMessageRecordBo bo) throws Exception{
+        boolean flag = sendMessageRecordService.updateByBo(bo);
+        if (!flag){
+            throw new ServiceException("消息发送记录修改失败");
+        }
+    }
+}

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

@@ -50,7 +50,7 @@ public class SyncRemoteXfTermService {
         Collection<Long> ids = JSONUtil.parseArray(msg).toBean(Collection.class);
         boolean flag = xfTermService.deleteWithValidByIds(ids,false);
         if (!flag){
-            throw new ServiceException("区域删除失败");
+            throw new ServiceException("设备删除失败");
         }
     }
 

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

@@ -19,3 +19,5 @@ org.dromara.server.base.service.yktOperation.SyncRemoteXfTermService
 org.dromara.server.base.service.yktOperation.SyncRemoteXfDiscountService
 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

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

@@ -2,11 +2,12 @@ package org.dromara.server.mq.consumer;
 
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
-import com.alibaba.nacos.api.config.annotation.NacosValue;
 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;
@@ -29,8 +30,11 @@ import static org.dromara.common.message.kafka.constant.KafkaTopicConstants.SYNC
 @Slf4j
 @Component
 public class KafkaConsumer {
+
     private final EventStrategyContext eventStrategyContext;
     private final YktEventStrategyContext yktEventStrategyContext;
+    private final SyncRemoteSendMessageRecordService syncRemoteSendMessageRecordService;
+
     @Value("${spring.system.tenantId}")
     private String tenantId;
 
@@ -54,21 +58,33 @@ public class KafkaConsumer {
      */
     @KafkaListener(topics = SYNC_DATA_TOPIC, groupId = "YTK_${spring.system.tenantId}")
     public void cloudOperationSync(ConsumerRecord<String, String> record){
+        KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
         try{
-            KafkaMessage<?> receiveMsg = JSONUtil.toBean(record.value(), KafkaMessage.class);
-            log.info("[接收到Kafka消息]-[{}]", receiveMsg);
-            String tenantId = receiveMsg.getHeader().getTenantId();
-            if(!tenantId.equals(this.tenantId)){
-                log.info("消息所属租户不匹配");
-                return;
+            if(receiveMsg.getHeader().getTenantId().equals(this.tenantId)){
+                String eventType = receiveMsg.getHeader().getEventType();
+                String sender = receiveMsg.getHeader().getSender();
+                yktEventStrategyContext.doMsgHandle(sender, eventType, receiveMsg.getBody());
+                syncRemoteSendMessageRecordService.editSendMessageRecord(initBo(receiveMsg.getHeader().getEventId(), "Y"));
             }
-            String eventType = receiveMsg.getHeader().getEventType();
-            String sender = receiveMsg.getHeader().getSender();
-
-            yktEventStrategyContext.doMsgHandle(sender, eventType, receiveMsg.getBody());
         } catch (Exception e){
+            try {
+                syncRemoteSendMessageRecordService.editSendMessageRecord(initBo(receiveMsg.getHeader().getEventId(), "N"));
+            } catch (Exception ex) {
+                log.error("消息发送记录更新失败", e);
+            }
             log.error("[kafka消息处理失败]-[消息:{}-[错误:{}]", record.value(), e.getMessage(), e);
-
         }
     }
+
+    /**
+     * 初始化消息记录Bo
+     * @param eventId 事件id
+     * @param consumeStatus 消费状态
+     */
+    private static RemoteSendMessageRecordBo initBo(String eventId, String consumeStatus){
+        RemoteSendMessageRecordBo bo = new RemoteSendMessageRecordBo();
+        bo.setEventId(eventId);
+        bo.setConsumeStatus(consumeStatus);
+        return bo;
+    }
 }

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

@@ -18,6 +18,9 @@ public class YktEventStrategyContext {
         this.strategyMap.putAll(strategyMap);
     }
 
+    /**
+     * 处理一卡通内部操作事件
+     */
     public void doMsgHandle(String sender, String eventType,Object msg) throws Exception{
         strategyMap.get(sender).doMsgHandle(eventType, msg);
     }

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

@@ -0,0 +1,33 @@
+package org.dromara.server.mq.event.kafka.impl.yktOperation;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.dromara.common.message.kafka.constant.MessageEventTypeConstants;
+import org.dromara.server.base.service.yktOperation.SyncRemotePtParameterService;
+import org.dromara.server.base.service.yktOperation.SyncRemotePtSubsidyService;
+import org.dromara.server.mq.event.kafka.IYktEventStrategy;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author Hz
+ * @date 2024/10/29
+ * @description 补助设置功能同步策略
+ */
+@Slf4j
+@RequiredArgsConstructor
+@Service(MessageEventTypeConstants.SUBSIDY_SENDER)
+public class PtSubsidyEventStrategyImpl implements IYktEventStrategy {
+
+    private final SyncRemotePtSubsidyService subsidyService;
+    @Override
+    public void doMsgHandle(String eventType, Object msg) throws Exception{
+        switch (eventType) {
+            case MessageEventTypeConstants.SUBSIDY_ADD -> subsidyService.addPtSubsidy(msg);
+            case MessageEventTypeConstants.SUBSIDY_EDIT -> subsidyService.editPtSubsidy(msg);
+            case MessageEventTypeConstants.SUBSIDY_REMOVE -> subsidyService.delPtSubsidy(msg);
+            case MessageEventTypeConstants.SUBSIDY_ITEM_REMOVE -> subsidyService.delPtSubsidyItem(msg);
+            default -> log.info("未知事件");
+        }
+    }
+
+}