feat(databus): 完成阶段四-DataBus Server完整功能

- 补充缺失的 API 类（DatabusMessage、DatabusBatchMessage、DatabusEventType） - 新增变更消息消费者（3个：部门、用户、岗位） - 新增数据提供者（3个：部门、用户、岗位） - 确认分发器服务（核心定向推送逻辑） - 确认全量同步与消息推送组件 - 确认管理后台 API（5个 Controller） - 确认 Service ��（4个核心服务） - 确认 DAL 层（7个 DO + Mapper） - 添加 databus-server starter 依赖到 pom.xml - 编译验证通过 Ref: docs/databus/implementation-checklist.md 任务 39-70
2025-12-01 23:44:32 +08:00
parent acdc73999a
commit bd90ec9d70
109 changed files with 7673 additions and 0 deletions
--- a/zt-module-databus/zt-module-databus-server/pom.xml
+++ b/zt-module-databus/zt-module-databus-server/pom.xml
@@ -42,6 +42,13 @@
            <version>${revision}</version>
        </dependency>

+        <!-- DataBus Server Starter -->
+        <dependency>
+            <groupId>com.zt.plat</groupId>
+            <artifactId>zt-spring-boot-starter-databus-server</artifactId>
+            <version>${revision}</version>
+        </dependency>
+
        <!-- 业务组件 -->
        <dependency>
            <groupId>com.zt.plat</groupId>
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusDeptChangeConsumer.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusDeptChangeConsumer.java
@@ -0,0 +1,90 @@
+package com.zt.plat.module.databus.mq.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.zt.plat.framework.databus.server.core.event.DatabusEvent;
+import com.zt.plat.framework.databus.server.core.sync.DatabusIncrementalSyncService;
+import com.zt.plat.module.system.api.mq.DatabusDeptChangeMessage;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Databus 部门变更消息消费者
+ * <p>
+ * 消费来自 system-server 的部门变更消息，通过增量同步服务进行：
+ * 1. 三态判断（事件/客户端/订阅是否启用）
+ * 2. 记录到 event_record 流水表
+ * 3. 推送到客户端专属 Topic（databus-sync-{clientCode}）
+ *
+ * @author ZT
+ */
+@Slf4j
+@Component
+@RocketMQMessageListener(
+        topic = DatabusDeptChangeMessage.TOPIC,
+        consumerGroup = DatabusDeptChangeMessage.TOPIC + "_CONSUMER"
+)
+public class DatabusDeptChangeConsumer implements RocketMQListener<DatabusDeptChangeMessage> {
+
+    private static final String SOURCE_SERVICE = "system-server";
+
+    @Resource
+    private DatabusIncrementalSyncService databusIncrementalSyncService;
+
+    @Resource
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void onMessage(DatabusDeptChangeMessage message) {
+        log.info("[Databus] 收到部门变更消息, action={}, deptId={}", message.getAction(), message.getDeptId());
+
+        try {
+            // 构建完整的业务数据快照
+            Map<String, Object> dataMap = new HashMap<>();
+            dataMap.put("id", message.getDeptId());
+            dataMap.put("code", message.getDeptCode());
+            dataMap.put("name", message.getDeptName());
+            dataMap.put("shortName", message.getShortName());
+            dataMap.put("parentId", message.getParentId());
+            dataMap.put("sort", message.getSort());
+            dataMap.put("leaderUserId", message.getLeaderUserId());
+            dataMap.put("phone", message.getPhone());
+            dataMap.put("email", message.getStatus());
+            dataMap.put("isCompany", message.getIsCompany());
+            dataMap.put("isGroup", message.getIsGroup());
+            dataMap.put("deptSource", message.getDeptSource());
+            dataMap.put("tenantId", message.getTenantId());
+            dataMap.put("eventTime", message.getEventTime());
+
+            // 构建完整的事件类型: system-dept-{action}
+            String eventType = String.format("system-dept-%s", message.getAction().toLowerCase());
+
+            // 构建 Databus 事件
+            DatabusEvent databusEvent = DatabusEvent.builder()
+                    .eventType(eventType)
+                    .eventAction(message.getAction())
+                    .dataSnapshot(objectMapper.writeValueAsString(dataMap))
+                    .dataVersion(1)
+                    .sourceService(SOURCE_SERVICE)
+                    .sourceTopic(DatabusDeptChangeMessage.TOPIC)
+                    .tenantId(message.getTenantId())
+                    .eventTime(message.getEventTime())
+                    .build();
+
+            // 调用增量同步服务处理（三态判断 + 记录流水 + 推送客户端Topic）
+            databusIncrementalSyncService.processEvent(databusEvent);
+
+            log.info("[Databus] 部门变更事件处理完成, eventType={}, deptId={}",
+                    eventType, message.getDeptId());
+        } catch (Exception e) {
+            log.error("[Databus] 处理部门变更消息失败, action={}, deptId={}",
+                    message.getAction(), message.getDeptId(), e);
+            throw new RuntimeException("处理部门变更消息失败", e);
+        }
+    }
+}
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusPostChangeConsumer.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusPostChangeConsumer.java
@@ -0,0 +1,73 @@
+package com.zt.plat.module.databus.mq.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.zt.plat.framework.databus.server.core.event.DatabusEvent;
+import com.zt.plat.framework.databus.server.core.sync.DatabusIncrementalSyncService;
+import com.zt.plat.module.system.api.mq.DatabusPostChangeMessage;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Databus 岗位变更消息消费者
+ */
+@Slf4j
+@Component
+@RocketMQMessageListener(
+        topic = DatabusPostChangeMessage.TOPIC,
+        consumerGroup = DatabusPostChangeMessage.TOPIC + "_CONSUMER"
+)
+public class DatabusPostChangeConsumer implements RocketMQListener<DatabusPostChangeMessage> {
+
+    private static final String SOURCE_SERVICE = "system-server";
+
+    @Resource
+    private DatabusIncrementalSyncService databusIncrementalSyncService;
+
+    @Resource
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void onMessage(DatabusPostChangeMessage message) {
+        log.info("[Databus] 收到岗位变更消息, action={}, postId={}", message.getAction(), message.getPostId());
+
+        try {
+            Map<String, Object> dataMap = new HashMap<>();
+            dataMap.put("id", message.getPostId());
+            dataMap.put("code", message.getPostCode());
+            dataMap.put("name", message.getPostName());
+            dataMap.put("sort", message.getSort());
+            dataMap.put("status", message.getStatus());
+            dataMap.put("remark", message.getRemark());
+            dataMap.put("tenantId", message.getTenantId());
+            dataMap.put("eventTime", message.getEventTime());
+
+            String eventType = String.format("system-post-%s", message.getAction().toLowerCase());
+
+            DatabusEvent databusEvent = DatabusEvent.builder()
+                    .eventType(eventType)
+                    .eventAction(message.getAction())
+                    .dataSnapshot(objectMapper.writeValueAsString(dataMap))
+                    .dataVersion(1)
+                    .sourceService(SOURCE_SERVICE)
+                    .sourceTopic(DatabusPostChangeMessage.TOPIC)
+                    .tenantId(message.getTenantId())
+                    .eventTime(message.getEventTime())
+                    .build();
+
+            databusIncrementalSyncService.processEvent(databusEvent);
+
+            log.info("[Databus] 岗位变更事件处理完成, eventType={}, postId={}",
+                    eventType, message.getPostId());
+        } catch (Exception e) {
+            log.error("[Databus] 处理岗位变更消息失败, action={}, postId={}",
+                    message.getAction(), message.getPostId(), e);
+            throw new RuntimeException("处理岗位变更消息失败", e);
+        }
+    }
+}
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusUserChangeConsumer.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/mq/consumer/DatabusUserChangeConsumer.java
@@ -0,0 +1,79 @@
+package com.zt.plat.module.databus.mq.consumer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.zt.plat.framework.databus.server.core.event.DatabusEvent;
+import com.zt.plat.framework.databus.server.core.sync.DatabusIncrementalSyncService;
+import com.zt.plat.module.system.api.mq.DatabusUserChangeMessage;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Databus 用户变更消息消费者
+ */
+@Slf4j
+@Component
+@RocketMQMessageListener(
+        topic = DatabusUserChangeMessage.TOPIC,
+        consumerGroup = DatabusUserChangeMessage.TOPIC + "_CONSUMER"
+)
+public class DatabusUserChangeConsumer implements RocketMQListener<DatabusUserChangeMessage> {
+
+    private static final String SOURCE_SERVICE = "system-server";
+
+    @Resource
+    private DatabusIncrementalSyncService databusIncrementalSyncService;
+
+    @Resource
+    private ObjectMapper objectMapper;
+
+    @Override
+    public void onMessage(DatabusUserChangeMessage message) {
+        log.info("[Databus] 收到用户变更消息, action={}, userId={}", message.getAction(), message.getUserId());
+
+        try {
+            Map<String, Object> dataMap = new HashMap<>();
+            dataMap.put("id", message.getUserId());
+            dataMap.put("username", message.getUsername());
+            dataMap.put("nickname", message.getNickname());
+            dataMap.put("remark", message.getRemark());
+            dataMap.put("deptIds", message.getDeptIds());
+            dataMap.put("postIds", message.getPostIds());
+            dataMap.put("email", message.getEmail());
+            dataMap.put("mobile", message.getMobile());
+            dataMap.put("sex", message.getSex());
+            dataMap.put("avatar", message.getAvatar());
+            dataMap.put("status", message.getStatus());
+            dataMap.put("userSource", message.getUserSource());
+            dataMap.put("tenantId", message.getTenantId());
+            dataMap.put("eventTime", message.getEventTime());
+
+            String eventType = String.format("system-user-%s", message.getAction().toLowerCase());
+
+            DatabusEvent databusEvent = DatabusEvent.builder()
+                    .eventType(eventType)
+                    .eventAction(message.getAction())
+                    .dataSnapshot(objectMapper.writeValueAsString(dataMap))
+                    .dataVersion(1)
+                    .sourceService(SOURCE_SERVICE)
+                    .sourceTopic(DatabusUserChangeMessage.TOPIC)
+                    .tenantId(message.getTenantId())
+                    .eventTime(message.getEventTime())
+                    .build();
+
+            databusIncrementalSyncService.processEvent(databusEvent);
+
+            log.info("[Databus] 用户变更事件处理完成, eventType={}, userId={}",
+                    eventType, message.getUserId());
+        } catch (Exception e) {
+            log.error("[Databus] 处理用户变更消息失败, action={}, userId={}",
+                    message.getAction(), message.getUserId(), e);
+            throw new RuntimeException("处理用户变更消息失败", e);
+        }
+    }
+}
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/DeptDataFeignProvider.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/DeptDataFeignProvider.java
@@ -0,0 +1,85 @@
+package com.zt.plat.module.databus.provider;
+
+import com.zt.plat.framework.common.pojo.CommonResult;
+import com.zt.plat.framework.databus.server.core.provider.DataProvider;
+import com.zt.plat.framework.databus.server.core.provider.DataProviderRegistry;
+import com.zt.plat.module.databus.api.dto.CursorPageReqDTO;
+import com.zt.plat.module.databus.api.dto.CursorPageResult;
+import com.zt.plat.module.databus.api.data.DatabusDeptData;
+import com.zt.plat.module.databus.api.provider.DatabusDeptProviderApi;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+/**
+ * 部门数据提供者
+ * <p>
+ * 通过 Feign 调用 system-server 获取部门数据
+ *
+ * @author ZT
+ */
+@Slf4j
+@Component
+public class DeptDataFeignProvider implements DataProvider<DatabusDeptData> {
+
+    public static final String PROVIDER_TYPE = "DEPT";
+
+    @Resource
+    private DatabusDeptProviderApi deptProviderApi;
+
+    @Resource
+    private DataProviderRegistry dataProviderRegistry;
+
+    @PostConstruct
+    public void init() {
+        dataProviderRegistry.register(this);
+    }
+
+    @Override
+    public String getProviderType() {
+        return PROVIDER_TYPE;
+    }
+
+    @Override
+    public CursorPageData<DatabusDeptData> getPageByCursor(LocalDateTime cursorTime, Long cursorId,
+                                                            int batchSize, Long tenantId) {
+        CursorPageReqDTO reqDTO = CursorPageReqDTO.builder()
+                .cursorTime(cursorTime)
+                .cursorId(cursorId)
+                .batchSize(batchSize)
+                .tenantId(tenantId)
+                .build();
+
+        CommonResult<CursorPageResult<DatabusDeptData>> result = deptProviderApi.getPageByCursor(reqDTO);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取部门数据失败: " + result.getMsg());
+        }
+
+        CursorPageResult<DatabusDeptData> pageResult = result.getData();
+        return CursorPageData.of(
+                pageResult.getList(),
+                pageResult.getNextCursorTime(),
+                pageResult.getNextCursorId(),
+                pageResult.getCount(),
+                Boolean.TRUE.equals(pageResult.getHasMore()),
+                (pageResult.getTotal() != null ? pageResult.getTotal() : 0L)
+        );
+    }
+
+    @Override
+    public long count(Long tenantId) {
+        CommonResult<Long> result = deptProviderApi.count(tenantId);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取部门总数失败: " + result.getMsg());
+        }
+        return result.getData();
+    }
+
+    @Override
+    public Long extractUid(DatabusDeptData data) {
+        return data.getId();
+    }
+}
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/PostDataFeignProvider.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/PostDataFeignProvider.java
@@ -0,0 +1,81 @@
+package com.zt.plat.module.databus.provider;
+
+import com.zt.plat.framework.common.pojo.CommonResult;
+import com.zt.plat.framework.databus.server.core.provider.DataProvider;
+import com.zt.plat.framework.databus.server.core.provider.DataProviderRegistry;
+import com.zt.plat.module.databus.api.dto.CursorPageReqDTO;
+import com.zt.plat.module.databus.api.dto.CursorPageResult;
+import com.zt.plat.module.databus.api.data.DatabusPostData;
+import com.zt.plat.module.databus.api.provider.DatabusPostProviderApi;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+/**
+ * 岗位数据提供者
+ */
+@Slf4j
+@Component
+public class PostDataFeignProvider implements DataProvider<DatabusPostData> {
+
+    public static final String PROVIDER_TYPE = "POST";
+
+    @Resource
+    private DatabusPostProviderApi postProviderApi;
+
+    @Resource
+    private DataProviderRegistry dataProviderRegistry;
+
+    @PostConstruct
+    public void init() {
+        dataProviderRegistry.register(this);
+    }
+
+    @Override
+    public String getProviderType() {
+        return PROVIDER_TYPE;
+    }
+
+    @Override
+    public CursorPageData<DatabusPostData> getPageByCursor(LocalDateTime cursorTime, Long cursorId,
+                                                             int batchSize, Long tenantId) {
+        CursorPageReqDTO reqDTO = CursorPageReqDTO.builder()
+                .cursorTime(cursorTime)
+                .cursorId(cursorId)
+                .batchSize(batchSize)
+                .tenantId(tenantId)
+                .build();
+
+        CommonResult<CursorPageResult<DatabusPostData>> result = postProviderApi.getPageByCursor(reqDTO);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取岗位数据失败: " + result.getMsg());
+        }
+
+        CursorPageResult<DatabusPostData> pageResult = result.getData();
+        return CursorPageData.of(
+                pageResult.getList(),
+                pageResult.getNextCursorTime(),
+                pageResult.getNextCursorId(),
+                pageResult.getCount(),
+                Boolean.TRUE.equals(pageResult.getHasMore()),
+                (pageResult.getTotal() != null ? pageResult.getTotal() : 0L)
+        );
+    }
+
+    @Override
+    public long count(Long tenantId) {
+        CommonResult<Long> result = postProviderApi.count(tenantId);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取岗位总数失败: " + result.getMsg());
+        }
+        return result.getData();
+    }
+
+    @Override
+    public Long extractUid(DatabusPostData data) {
+        return data.getId();
+    }
+}
--- a/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/UserDataFeignProvider.java
+++ b/zt-module-databus/zt-module-databus-server/src/main/java/com/zt/plat/module/databus/provider/UserDataFeignProvider.java
@@ -0,0 +1,81 @@
+package com.zt.plat.module.databus.provider;
+
+import com.zt.plat.framework.common.pojo.CommonResult;
+import com.zt.plat.framework.databus.server.core.provider.DataProvider;
+import com.zt.plat.framework.databus.server.core.provider.DataProviderRegistry;
+import com.zt.plat.module.databus.api.dto.CursorPageReqDTO;
+import com.zt.plat.module.databus.api.dto.CursorPageResult;
+import com.zt.plat.module.databus.api.data.DatabusAdminUserData;
+import com.zt.plat.module.databus.api.provider.DatabusUserProviderApi;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import java.time.LocalDateTime;
+
+/**
+ * 用户数据提供者
+ */
+@Slf4j
+@Component
+public class UserDataFeignProvider implements DataProvider<DatabusAdminUserData> {
+
+    public static final String PROVIDER_TYPE = "USER";
+
+    @Resource
+    private DatabusUserProviderApi userProviderApi;
+
+    @Resource
+    private DataProviderRegistry dataProviderRegistry;
+
+    @PostConstruct
+    public void init() {
+        dataProviderRegistry.register(this);
+    }
+
+    @Override
+    public String getProviderType() {
+        return PROVIDER_TYPE;
+    }
+
+    @Override
+    public CursorPageData<DatabusAdminUserData> getPageByCursor(LocalDateTime cursorTime, Long cursorId,
+                                                                  int batchSize, Long tenantId) {
+        CursorPageReqDTO reqDTO = CursorPageReqDTO.builder()
+                .cursorTime(cursorTime)
+                .cursorId(cursorId)
+                .batchSize(batchSize)
+                .tenantId(tenantId)
+                .build();
+
+        CommonResult<CursorPageResult<DatabusAdminUserData>> result = userProviderApi.getPageByCursor(reqDTO);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取用户数据失败: " + result.getMsg());
+        }
+
+        CursorPageResult<DatabusAdminUserData> pageResult = result.getData();
+        return CursorPageData.of(
+                pageResult.getList(),
+                pageResult.getNextCursorTime(),
+                pageResult.getNextCursorId(),
+                pageResult.getCount(),
+                Boolean.TRUE.equals(pageResult.getHasMore()),
+                (pageResult.getTotal() != null ? pageResult.getTotal() : 0L)
+        );
+    }
+
+    @Override
+    public long count(Long tenantId) {
+        CommonResult<Long> result = userProviderApi.count(tenantId);
+        if (!result.isSuccess()) {
+            throw new RuntimeException("获取用户总数失败: " + result.getMsg());
+        }
+        return result.getData();
+    }
+
+    @Override
+    public Long extractUid(DatabusAdminUserData data) {
+        return data.getId();
+    }
+}