zgty/zt-dsc
9
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update：数据同步分发机构岗位绑定关系

Browse Source
This commit is contained in:
hewencai
2025-12-16 12:03:15 +08:00
parent df3e80d907
commit ba1bc1fb6f
31 changed files with 1278 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/core/listener/DatabusConsumerRegistry.java
View File

@@ -1,148 +0,0 @@
package com.zt.plat.framework.databus.client.core.listener;
import com.zt.plat.framework.databus.client.config.DatabusSyncClientProperties;
import com.zt.plat.framework.databus.client.core.processor.MessageProcessor;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
/**
* Databus Consumer 注册器
* <p>
* 根据 DatabusEventType 枚举自动为所有事件类型创建消费者
* Topic格式: {topicBase}-{module}-{entity}-{action}-{clientCode}
* 例如: databus-sync-system-user-create-company-a
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client.mq", name = "enabled", havingValue = "true", matchIfMissing = true)
public class DatabusConsumerRegistry {
private final MessageProcessor messageProcessor;
private final DatabusSyncClientProperties properties;
/**
* 管理的消费者列表
*/
private final List<DefaultMQPushConsumer> consumers = new ArrayList<>();
public DatabusConsumerRegistry(MessageProcessor messageProcessor, DatabusSyncClientProperties properties) {
this.messageProcessor = messageProcessor;
this.properties = properties;
}
@PostConstruct
public void init() {
if (!Boolean.TRUE.equals(properties.getEnabled())) {
log.info("[Databus Client] 客户端未启用，跳过初始化");
return;
}
String nameServer = properties.getMq().getNameServer();
if (nameServer == null || nameServer.isEmpty()) {
log.warn("[Databus Client] RocketMQ nameServer未配置，跳过MQ消费者初始化");
return;
}
String clientCode = properties.getClientCode();
if (clientCode == null || clientCode.isEmpty()) {
log.error("[Databus Client] clientCode未配置，无法订阅Topic");
return;
}
String topicBase = properties.getMq().getTopicBase();
String consumerGroupPrefix = properties.getMq().getConsumerGroupPrefix();
if (consumerGroupPrefix == null || consumerGroupPrefix.isEmpty()) {
consumerGroupPrefix = "databus-client-" + clientCode;
}
// 为每个事件类型创建独立的消费者
for (DatabusEventType eventType : DatabusEventType.values()) {
try {
createConsumer(eventType, topicBase, clientCode, consumerGroupPrefix, nameServer);
} catch (Exception e) {
log.error("[Databus Client] 创建消费者失败, eventType={}", eventType.name(), e);
}
}
log.info("[Databus Client] 消费者注册完成，共创建 {} 个消费者", consumers.size());
}
/**
* 为指定事件类型创建消费者
*/
private void createConsumer(DatabusEventType eventType, String topicBase, String clientCode, String consumerGroupPrefix, String nameServer) throws MQClientException {
// Topic: databus-sync-system-user-create-company-a
String topic = eventType.getTopic(topicBase, clientCode);
// ConsumerGroup: databus-client-company-a-system-user-create
String consumerGroup = String.format("%s-%s", consumerGroupPrefix, eventType.getTopicSuffix());
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
consumer.setNamesrvAddr(nameServer);
consumer.setConsumeThreadMin(properties.getMq().getConsumeThreadMin());
consumer.setConsumeThreadMax(properties.getMq().getConsumeThreadMax());
consumer.setMaxReconsumeTimes(properties.getMq().getMaxReconsumeTimes());
// 订阅Topic
consumer.subscribe(topic, "*");
// 设置消息监听器
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
for (MessageExt msg : msgs) {
try {
String messageBody = new String(msg.getBody(), StandardCharsets.UTF_8);
log.info("[Databus Client] 收到消息, topic={}, msgId={}, eventType={}",
msg.getTopic(), msg.getMsgId(), eventType.name());
messageProcessor.process(messageBody, eventType);
} catch (Exception e) {
log.error("[Databus Client] 消息处理失败, topic={}, msgId={}, reconsumeTimes={}",
msg.getTopic(), msg.getMsgId(), msg.getReconsumeTimes(), e);
if (msg.getReconsumeTimes() >= properties.getMq().getMaxReconsumeTimes()) {
log.error("[Databus Client] 重试次数已达上限，放弃处理, msgId={}", msg.getMsgId());
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();
consumers.add(consumer);
log.info("[Databus Client] 消费者启动成功, topic={}, consumerGroup={}, event={}",
topic, consumerGroup, eventType.getName());
}
@PreDestroy
public void destroy() {
for (DefaultMQPushConsumer consumer : consumers) {
try {
consumer.shutdown();
} catch (Exception e) {
log.error("[Databus Client] 关闭消费者失败", e);
}
}
consumers.clear();
log.info("[Databus Client] 所有消费者已关闭");
}
}

12
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/core/registry/HandlerRegistry.java
View File

@@ -66,13 +66,15 @@ public class HandlerRegistry {
public void init() {
log.info("[HandlerRegistry] 开始初始化 Handler 注册表...");
// 注册增量同步 Handler
// 注册增量同步 Handler（支持一个Handler注册多个事件类型）
if (syncEventHandlers != null) {
for (SyncEventHandler<?> handler : syncEventHandlers) {
DatabusEventType eventType = handler.getSupportedEventType();
incrementalHandlers.put(eventType, handler);
log.info("[HandlerRegistry] 注册增量Handler: {} -> {}",
eventType, handler.getClass().getSimpleName());
List<DatabusEventType> eventTypes = handler.getSupportedEventTypes();
for (DatabusEventType eventType : eventTypes) {
incrementalHandlers.put(eventType, handler);
log.info("[HandlerRegistry] 注册增量Handler: {} -> {}",
eventType, handler.getClass().getSimpleName());
}
}
}

31
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/SyncEventHandler.java
View File

@@ -5,22 +5,49 @@ import com.zt.plat.module.databus.enums.DatabusEventType;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.List;
/**
* 同步事件处理器接口
* <p>
* 业务系统需要实现此接口来处理增量数据同步
* <p>
* 支持单事件和多事件处理：
* - 单事件：实现 getSupportedEventType()
* - 多事件：实现 getSupportedEventTypes()（推荐，一个Handler处理多个操作）
*
* @author ZT
*/
public interface SyncEventHandler<T> {
/**
* 获取支持的事件类型
* 获取支持的事件类型（单个）
* <p>
* 默认实现：返回 getSupportedEventTypes() 的第一个元素
* <p>
* 建议：新Handler优先实现 getSupportedEventTypes()
*
* @return 事件类型枚举
*/
DatabusEventType getSupportedEventType();
default DatabusEventType getSupportedEventType() {
List<DatabusEventType> types = getSupportedEventTypes();
return types.isEmpty() ? null : types.get(0);
}
/**
* 获取支持的事件类型列表（多个）
* <p>
* 默认实现：返回包含 getSupportedEventType() 的单元素列表
* <p>
* 子类可以覆盖此方法返回多个事件类型，从而一个Handler处理多个操作
*
* @return 事件类型列表
*/
default List<DatabusEventType> getSupportedEventTypes() {
DatabusEventType type = getSupportedEventType();
return type == null ? Collections.emptyList() : Collections.singletonList(type);
}
/**
* 处理同步消息

50
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/dept/SystemDeptCreateHandler.java
View File

@@ -1,50 +0,0 @@
package com.zt.plat.framework.databus.client.handler.dept;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusDeptData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 部门创建事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 DeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(DeptSyncService.class)
public class SystemDeptCreateHandler implements SyncEventHandler<DatabusDeptData> {
@Resource
private DeptSyncService deptSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_DEPT_CREATE;
}
@Override
public void handle(DatabusMessage<DatabusDeptData> message) {
DatabusDeptData data = message.getData();
log.info("[DeptSync] 收到部门创建事件, id={}, name={}, parentId={}",
data.getId(), data.getName(), data.getParentId());
try {
deptSyncService.create(data);
log.info("[DeptSync] 部门创建成功, id={}", data.getId());
} catch (Exception e) {
log.error("[DeptSync] 部门创建失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/dept/SystemDeptDeleteHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.dept;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusDeptData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 部门删除事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 DeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(DeptSyncService.class)
public class SystemDeptDeleteHandler implements SyncEventHandler<DatabusDeptData> {
@Resource
private DeptSyncService deptSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_DEPT_DELETE;
}
@Override
public void handle(DatabusMessage<DatabusDeptData> message) {
DatabusDeptData data = message.getData();
log.info("[DeptSync] 收到部门删除事件, id={}", data.getId());
try {
deptSyncService.delete(data.getId());
log.info("[DeptSync] 部门删除成功, id={}", data.getId());
} catch (Exception e) {
log.error("[DeptSync] 部门删除失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

81
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/dept/SystemDeptHandler.java Normal file
View File

@@ -0,0 +1,81 @@
package com.zt.plat.framework.databus.client.handler.dept;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusDeptData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 部门增量同步事件处理器（合并版）
* <p>
* 处理事件类型：
* - SYSTEM_DEPT_CREATE：部门创建
* - SYSTEM_DEPT_UPDATE：部门更新
* - SYSTEM_DEPT_DELETE：部门删除
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 DeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(DeptSyncService.class)
public class SystemDeptHandler implements SyncEventHandler<DatabusDeptData> {
@Resource
private DeptSyncService deptSyncService;
@Override
public List<DatabusEventType> getSupportedEventTypes() {
return Arrays.asList(
DatabusEventType.SYSTEM_DEPT_CREATE,
DatabusEventType.SYSTEM_DEPT_UPDATE,
DatabusEventType.SYSTEM_DEPT_DELETE
);
}
@Override
public void handle(DatabusMessage<DatabusDeptData> message) {
DatabusDeptData data = message.getData();
DatabusEventType eventType = message.getEventType();
log.info("[DeptSync] 收到部门{}事件, id={}, name={}, parentId={}",
eventType.getAction(), data.getId(), data.getName(), data.getParentId());
try {
switch (eventType) {
case SYSTEM_DEPT_CREATE:
deptSyncService.create(data);
log.info("[DeptSync] 部门创建成功, id={}", data.getId());
break;
case SYSTEM_DEPT_UPDATE:
deptSyncService.update(data);
log.info("[DeptSync] 部门更新成功, id={}", data.getId());
break;
case SYSTEM_DEPT_DELETE:
deptSyncService.delete(data.getId());
log.info("[DeptSync] 部门删除成功, id={}", data.getId());
break;
default:
log.warn("[DeptSync] 未知事件类型: {}", eventType);
}
} catch (Exception e) {
log.error("[DeptSync] 部门{}失败, id={}", eventType.getAction(), data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/dept/SystemDeptUpdateHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.dept;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusDeptData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 部门更新事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 DeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(DeptSyncService.class)
public class SystemDeptUpdateHandler implements SyncEventHandler<DatabusDeptData> {
@Resource
private DeptSyncService deptSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_DEPT_UPDATE;
}
@Override
public void handle(DatabusMessage<DatabusDeptData> message) {
DatabusDeptData data = message.getData();
log.info("[DeptSync] 收到部门更新事件, id={}, name={}", data.getId(), data.getName());
try {
deptSyncService.update(data);
log.info("[DeptSync] 部门更新成功, id={}", data.getId());
} catch (Exception e) {
log.error("[DeptSync] 部门更新失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

50
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/post/SystemPostCreateHandler.java
View File

@@ -1,50 +0,0 @@
package com.zt.plat.framework.databus.client.handler.post;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusPostData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 岗位创建事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 PostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(PostSyncService.class)
public class SystemPostCreateHandler implements SyncEventHandler<DatabusPostData> {
@Resource
private PostSyncService postSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_POST_CREATE;
}
@Override
public void handle(DatabusMessage<DatabusPostData> message) {
DatabusPostData data = message.getData();
log.info("[PostSync] 收到岗位创建事件, id={}, name={}, code={}",
data.getId(), data.getName(), data.getCode());
try {
postSyncService.create(data);
log.info("[PostSync] 岗位创建成功, id={}", data.getId());
} catch (Exception e) {
log.error("[PostSync] 岗位创建失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/post/SystemPostDeleteHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.post;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusPostData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 岗位删除事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 PostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(PostSyncService.class)
public class SystemPostDeleteHandler implements SyncEventHandler<DatabusPostData> {
@Resource
private PostSyncService postSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_POST_DELETE;
}
@Override
public void handle(DatabusMessage<DatabusPostData> message) {
DatabusPostData data = message.getData();
log.info("[PostSync] 收到岗位删除事件, id={}", data.getId());
try {
postSyncService.delete(data.getId());
log.info("[PostSync] 岗位删除成功, id={}", data.getId());
} catch (Exception e) {
log.error("[PostSync] 岗位删除失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

81
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/post/SystemPostHandler.java Normal file
View File

@@ -0,0 +1,81 @@
package com.zt.plat.framework.databus.client.handler.post;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusPostData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 岗位增量同步事件处理器（合并版）
* <p>
* 处理事件类型：
* - SYSTEM_POST_CREATE：岗位创建
* - SYSTEM_POST_UPDATE：岗位更新
* - SYSTEM_POST_DELETE：岗位删除
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 PostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(PostSyncService.class)
public class SystemPostHandler implements SyncEventHandler<DatabusPostData> {
@Resource
private PostSyncService postSyncService;
@Override
public List<DatabusEventType> getSupportedEventTypes() {
return Arrays.asList(
DatabusEventType.SYSTEM_POST_CREATE,
DatabusEventType.SYSTEM_POST_UPDATE,
DatabusEventType.SYSTEM_POST_DELETE
);
}
@Override
public void handle(DatabusMessage<DatabusPostData> message) {
DatabusPostData data = message.getData();
DatabusEventType eventType = message.getEventType();
log.info("[PostSync] 收到岗位{}事件, id={}, name={}, code={}",
eventType.getAction(), data.getId(), data.getName(), data.getCode());
try {
switch (eventType) {
case SYSTEM_POST_CREATE:
postSyncService.create(data);
log.info("[PostSync] 岗位创建成功, id={}", data.getId());
break;
case SYSTEM_POST_UPDATE:
postSyncService.update(data);
log.info("[PostSync] 岗位更新成功, id={}", data.getId());
break;
case SYSTEM_POST_DELETE:
postSyncService.delete(data.getId());
log.info("[PostSync] 岗位删除成功, id={}", data.getId());
break;
default:
log.warn("[PostSync] 未知事件类型: {}", eventType);
}
} catch (Exception e) {
log.error("[PostSync] 岗位{}失败, id={}", eventType.getAction(), data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

50
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/post/SystemPostUpdateHandler.java
View File

@@ -1,50 +0,0 @@
package com.zt.plat.framework.databus.client.handler.post;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusPostData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 岗位更新事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 PostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(PostSyncService.class)
public class SystemPostUpdateHandler implements SyncEventHandler<DatabusPostData> {
@Resource
private PostSyncService postSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_POST_UPDATE;
}
@Override
public void handle(DatabusMessage<DatabusPostData> message) {
DatabusPostData data = message.getData();
log.info("[PostSync] 收到岗位更新事件, id={}, name={}, code={}",
data.getId(), data.getName(), data.getCode());
try {
postSyncService.update(data);
log.info("[PostSync] 岗位更新成功, id={}", data.getId());
} catch (Exception e) {
log.error("[PostSync] 岗位更新失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/user/SystemUserCreateHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.user;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusAdminUserData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 用户创建事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 AdminUserSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(AdminUserSyncService.class)
public class SystemUserCreateHandler implements SyncEventHandler<DatabusAdminUserData> {
@Resource
private AdminUserSyncService adminUserSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_USER_CREATE;
}
@Override
public void handle(DatabusMessage<DatabusAdminUserData> message) {
DatabusAdminUserData data = message.getData();
log.info("[UserSync] 收到用户创建事件, id={}, username={}", data.getId(), data.getUsername());
try {
adminUserSyncService.create(data);
log.info("[UserSync] 用户创建成功, id={}", data.getId());
} catch (Exception e) {
log.error("[UserSync] 用户创建失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/user/SystemUserDeleteHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.user;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusAdminUserData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 用户删除事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 AdminUserSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(AdminUserSyncService.class)
public class SystemUserDeleteHandler implements SyncEventHandler<DatabusAdminUserData> {
@Resource
private AdminUserSyncService adminUserSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_USER_DELETE;
}
@Override
public void handle(DatabusMessage<DatabusAdminUserData> message) {
DatabusAdminUserData data = message.getData();
log.info("[UserSync] 收到用户删除事件, id={}", data.getId());
try {
adminUserSyncService.delete(data.getId());
log.info("[UserSync] 用户删除成功, id={}", data.getId());
} catch (Exception e) {
log.error("[UserSync] 用户删除失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

81
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/user/SystemUserHandler.java Normal file
View File

@@ -0,0 +1,81 @@
package com.zt.plat.framework.databus.client.handler.user;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusAdminUserData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 用户增量同步事件处理器（合并版）
* <p>
* 处理事件类型：
* - SYSTEM_USER_CREATE：用户创建
* - SYSTEM_USER_UPDATE：用户更新
* - SYSTEM_USER_DELETE：用户删除
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 AdminUserSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(AdminUserSyncService.class)
public class SystemUserHandler implements SyncEventHandler<DatabusAdminUserData> {
@Resource
private AdminUserSyncService adminUserSyncService;
@Override
public List<DatabusEventType> getSupportedEventTypes() {
return Arrays.asList(
DatabusEventType.SYSTEM_USER_CREATE,
DatabusEventType.SYSTEM_USER_UPDATE,
DatabusEventType.SYSTEM_USER_DELETE
);
}
@Override
public void handle(DatabusMessage<DatabusAdminUserData> message) {
DatabusAdminUserData data = message.getData();
DatabusEventType eventType = message.getEventType();
log.info("[UserSync] 收到用户{}事件, id={}, username={}, nickname={}",
eventType.getAction(), data.getId(), data.getUsername(), data.getNickname());
try {
switch (eventType) {
case SYSTEM_USER_CREATE:
adminUserSyncService.create(data);
log.info("[UserSync] 用户创建成功, id={}", data.getId());
break;
case SYSTEM_USER_UPDATE:
adminUserSyncService.update(data);
log.info("[UserSync] 用户更新成功, id={}", data.getId());
break;
case SYSTEM_USER_DELETE:
adminUserSyncService.delete(data.getId());
log.info("[UserSync] 用户删除成功, id={}", data.getId());
break;
default:
log.warn("[UserSync] 未知事件类型: {}", eventType);
}
} catch (Exception e) {
log.error("[UserSync] 用户{}失败, id={}", eventType.getAction(), data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

49
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/user/SystemUserUpdateHandler.java
View File

@@ -1,49 +0,0 @@
package com.zt.plat.framework.databus.client.handler.user;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusAdminUserData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 用户更新事件处理器
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 AdminUserSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(AdminUserSyncService.class)
public class SystemUserUpdateHandler implements SyncEventHandler<DatabusAdminUserData> {
@Resource
private AdminUserSyncService adminUserSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_USER_UPDATE;
}
@Override
public void handle(DatabusMessage<DatabusAdminUserData> message) {
DatabusAdminUserData data = message.getData();
log.info("[UserSync] 收到用户更新事件, id={}, username={}", data.getId(), data.getUsername());
try {
adminUserSyncService.update(data);
log.info("[UserSync] 用户更新成功, id={}", data.getId());
} catch (Exception e) {
log.error("[UserSync] 用户更新失败, id={}", data.getId(), e);
throw e; // 抛出异常触发重试
}
}
}

70
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userdept/SystemUserDeptFullHandler.java Normal file
View File

@@ -0,0 +1,70 @@
package com.zt.plat.framework.databus.client.handler.userdept;
import com.zt.plat.framework.databus.client.handler.BatchSyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusUserDeptData;
import com.zt.plat.module.databus.api.message.DatabusBatchMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 用户-部门关系全量同步事件处理器（批量处理）
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 UserDeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(UserDeptSyncService.class)
public class SystemUserDeptFullHandler implements BatchSyncEventHandler<DatabusUserDeptData> {
@Resource
private UserDeptSyncService userDeptSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_USER_DEPT_FULL;
}
@Override
public void onFullSyncStart(DatabusBatchMessage<DatabusUserDeptData> message) {
log.info("[UserDeptSync] 开始用户-部门关系全量同步, taskId={}, totalBatch={}",
message.getTaskId(), message.getTotalBatch());
}
@Override
public void handleBatch(DatabusBatchMessage<DatabusUserDeptData> message) {
log.info("[UserDeptSync] 处理用户-部门关系批次数据, taskId={}, batchNo={}/{}, size={}",
message.getTaskId(), message.getBatchNo(), message.getTotalBatch(),
message.getDataList().size());
// 逐条处理全量同步数据
for (DatabusUserDeptData data : message.getDataList()) {
try {
userDeptSyncService.fullSync(data);
log.debug("[UserDeptSync] 用户-部门关系全量同步成功, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
} catch (Exception e) {
log.error("[UserDeptSync] 用户-部门关系全量同步失败, userId={}, deptId={}",
data.getUserId(), data.getDeptId(), e);
// 单条失败不影响其他数据，继续处理
}
}
log.info("[UserDeptSync] 用户-部门关系批次处理完成, taskId={}, batchNo={}/{}",
message.getTaskId(), message.getBatchNo(), message.getTotalBatch());
}
@Override
public void onFullSyncComplete(DatabusBatchMessage<DatabusUserDeptData> message) {
log.info("[UserDeptSync] 用户-部门关系全量同步完成, taskId={}, totalBatch={}",
message.getTaskId(), message.getTotalBatch());
}
}

87
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userdept/SystemUserDeptHandler.java Normal file
View File

@@ -0,0 +1,87 @@
package com.zt.plat.framework.databus.client.handler.userdept;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusUserDeptData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 用户-部门关系增量同步事件处理器（合并版）
* <p>
* 处理事件类型：
* - SYSTEM_USER_DEPT_CREATE：用户-部门关系创建
* - SYSTEM_USER_DEPT_UPDATE：用户-部门关系更新
* - SYSTEM_USER_DEPT_DELETE：用户-部门关系删除
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 UserDeptSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(UserDeptSyncService.class)
public class SystemUserDeptHandler implements SyncEventHandler<DatabusUserDeptData> {
@Resource
private UserDeptSyncService userDeptSyncService;
@Override
public List<DatabusEventType> getSupportedEventTypes() {
return Arrays.asList(
DatabusEventType.SYSTEM_USER_DEPT_CREATE,
DatabusEventType.SYSTEM_USER_DEPT_UPDATE,
DatabusEventType.SYSTEM_USER_DEPT_DELETE
);
}
@Override
public void handle(DatabusMessage<DatabusUserDeptData> message) {
DatabusUserDeptData data = message.getData();
DatabusEventType eventType = message.getEventType();
log.info("[UserDeptSync] 收到用户-部门关系{}事件, userId={}, deptId={}",
eventType.getAction(), data != null ? data.getUserId() : null,
data != null ? data.getDeptId() : null);
try {
switch (eventType) {
case SYSTEM_USER_DEPT_CREATE:
userDeptSyncService.create(data);
log.info("[UserDeptSync] 用户-部门关系创建成功, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
break;
case SYSTEM_USER_DEPT_UPDATE:
userDeptSyncService.update(data);
log.info("[UserDeptSync] 用户-部门关系更新成功, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
break;
case SYSTEM_USER_DEPT_DELETE:
userDeptSyncService.delete(data.getId());
log.info("[UserDeptSync] 用户-部门关系删除成功, id={}", data.getId());
break;
default:
log.warn("[UserDeptSync] 未知事件类型: {}", eventType);
}
} catch (Exception e) {
log.error("[UserDeptSync] 用户-部门关系{}失败, userId={}, deptId={}",
eventType.getAction(),
data != null ? data.getUserId() : null,
data != null ? data.getDeptId() : null, e);
throw e; // 抛出异常触发重试
}
}
}

44
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userdept/UserDeptSyncService.java Normal file
View File

@@ -0,0 +1,44 @@
package com.zt.plat.framework.databus.client.handler.userdept;
import com.zt.plat.module.databus.api.data.DatabusUserDeptData;
/**
* 用户-部门关系同步服务接口
* <p>
* 分公司需要实现此接口，完成数据的本地持久化
* 或通过默认实现 {@link UserDeptSyncServiceImpl} 使用 Feign 调用远程 API
*
* @author ZT
*/
public interface UserDeptSyncService {
/**
* 创建用户-部门关系（增量同步）
*
* @param data 用户-部门关系数据
*/
void create(DatabusUserDeptData data);
/**
* 更新用户-部门关系（增量同步）
*
* @param data 用户-部门关系数据
*/
void update(DatabusUserDeptData data);
/**
* 删除用户-部门关系（增量同步）
*
* @param id 关系ID
*/
void delete(Long id);
/**
* 全量同步单条数据
* <p>
* 逻辑：存在则更新，不存在则插入
*
* @param data 用户-部门关系数据
*/
void fullSync(DatabusUserDeptData data);
}

56
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userdept/UserDeptSyncServiceImpl.java Normal file
View File

@@ -0,0 +1,56 @@
package com.zt.plat.framework.databus.client.handler.userdept;
import com.zt.plat.module.databus.api.data.DatabusUserDeptData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
/**
* 用户-部门关系同步服务实现
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* <p>
* 注意：由于用户-部门关系通常集成在用户管理中，此实现为占位符。
* 分公司可以根据实际情况：
* 1. 自定义实现此接口，直接操作本地数据库
* 2. 或者通过用户管理 API 间接处理关联关系
*
* @author ZT
*/
@Slf4j
@Service
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
public class UserDeptSyncServiceImpl implements UserDeptSyncService {
@Override
public void create(DatabusUserDeptData data) {
log.info("[UserDeptSync] 收到创建用户-部门关系请求, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
log.warn("[UserDeptSync] 用户-部门关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法，通过本地 API 或直接数据库操作完成同步
}
@Override
public void update(DatabusUserDeptData data) {
log.info("[UserDeptSync] 收到更新用户-部门关系请求, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
log.warn("[UserDeptSync] 用户-部门关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法
}
@Override
public void delete(Long id) {
log.info("[UserDeptSync] 收到删除用户-部门关系请求, id={}", id);
log.warn("[UserDeptSync] 用户-部门关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法
}
@Override
public void fullSync(DatabusUserDeptData data) {
log.info("[UserDeptSync] 收到全量同步用户-部门关系请求, userId={}, deptId={}",
data.getUserId(), data.getDeptId());
log.warn("[UserDeptSync] 用户-部门关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法，逻辑：存在则更新，不存在则插入
}
}

70
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userpost/SystemUserPostFullHandler.java Normal file
View File

@@ -0,0 +1,70 @@
package com.zt.plat.framework.databus.client.handler.userpost;
import com.zt.plat.framework.databus.client.handler.BatchSyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusUserPostData;
import com.zt.plat.module.databus.api.message.DatabusBatchMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 用户-岗位关系全量同步事件处理器（批量处理）
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 UserPostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(UserPostSyncService.class)
public class SystemUserPostFullHandler implements BatchSyncEventHandler<DatabusUserPostData> {
@Resource
private UserPostSyncService userPostSyncService;
@Override
public DatabusEventType getSupportedEventType() {
return DatabusEventType.SYSTEM_USER_POST_FULL;
}
@Override
public void onFullSyncStart(DatabusBatchMessage<DatabusUserPostData> message) {
log.info("[UserPostSync] 开始用户-岗位关系全量同步, taskId={}, totalBatch={}",
message.getTaskId(), message.getTotalBatch());
}
@Override
public void handleBatch(DatabusBatchMessage<DatabusUserPostData> message) {
log.info("[UserPostSync] 处理用户-岗位关系批次数据, taskId={}, batchNo={}/{}, size={}",
message.getTaskId(), message.getBatchNo(), message.getTotalBatch(),
message.getDataList().size());
// 逐条处理全量同步数据
for (DatabusUserPostData data : message.getDataList()) {
try {
userPostSyncService.fullSync(data);
log.debug("[UserPostSync] 用户-岗位关系全量同步成功, userId={}, postId={}",
data.getUserId(), data.getPostId());
} catch (Exception e) {
log.error("[UserPostSync] 用户-岗位关系全量同步失败, userId={}, postId={}",
data.getUserId(), data.getPostId(), e);
// 单条失败不影响其他数据，继续处理
}
}
log.info("[UserPostSync] 用户-岗位关系批次处理完成, taskId={}, batchNo={}/{}",
message.getTaskId(), message.getBatchNo(), message.getTotalBatch());
}
@Override
public void onFullSyncComplete(DatabusBatchMessage<DatabusUserPostData> message) {
log.info("[UserPostSync] 用户-岗位关系全量同步完成, taskId={}, totalBatch={}",
message.getTaskId(), message.getTotalBatch());
}
}

87
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userpost/SystemUserPostHandler.java Normal file
View File

@@ -0,0 +1,87 @@
package com.zt.plat.framework.databus.client.handler.userpost;
import com.zt.plat.framework.databus.client.handler.SyncEventHandler;
import com.zt.plat.module.databus.api.data.DatabusUserPostData;
import com.zt.plat.module.databus.api.message.DatabusMessage;
import com.zt.plat.module.databus.enums.DatabusEventType;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.Arrays;
import java.util.List;
/**
* 用户-岗位关系增量同步事件处理器（合并版）
* <p>
* 处理事件类型：
* - SYSTEM_USER_POST_CREATE：用户-岗位关系创建
* - SYSTEM_USER_POST_UPDATE：用户-岗位关系更新
* - SYSTEM_USER_POST_DELETE：用户-岗位关系删除
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* 2. 存在 UserPostSyncService Bean
*
* @author ZT
*/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
@ConditionalOnBean(UserPostSyncService.class)
public class SystemUserPostHandler implements SyncEventHandler<DatabusUserPostData> {
@Resource
private UserPostSyncService userPostSyncService;
@Override
public List<DatabusEventType> getSupportedEventTypes() {
return Arrays.asList(
DatabusEventType.SYSTEM_USER_POST_CREATE,
DatabusEventType.SYSTEM_USER_POST_UPDATE,
DatabusEventType.SYSTEM_USER_POST_DELETE
);
}
@Override
public void handle(DatabusMessage<DatabusUserPostData> message) {
DatabusUserPostData data = message.getData();
DatabusEventType eventType = message.getEventType();
log.info("[UserPostSync] 收到用户-岗位关系{}事件, userId={}, postId={}",
eventType.getAction(), data != null ? data.getUserId() : null,
data != null ? data.getPostId() : null);
try {
switch (eventType) {
case SYSTEM_USER_POST_CREATE:
userPostSyncService.create(data);
log.info("[UserPostSync] 用户-岗位关系创建成功, userId={}, postId={}",
data.getUserId(), data.getPostId());
break;
case SYSTEM_USER_POST_UPDATE:
userPostSyncService.update(data);
log.info("[UserPostSync] 用户-岗位关系更新成功, userId={}, postId={}",
data.getUserId(), data.getPostId());
break;
case SYSTEM_USER_POST_DELETE:
userPostSyncService.delete(data.getId());
log.info("[UserPostSync] 用户-岗位关系删除成功, id={}", data.getId());
break;
default:
log.warn("[UserPostSync] 未知事件类型: {}", eventType);
}
} catch (Exception e) {
log.error("[UserPostSync] 用户-岗位关系{}失败, userId={}, postId={}",
eventType.getAction(),
data != null ? data.getUserId() : null,
data != null ? data.getPostId() : null, e);
throw e; // 抛出异常触发重试
}
}
}

44
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userpost/UserPostSyncService.java Normal file
View File

@@ -0,0 +1,44 @@
package com.zt.plat.framework.databus.client.handler.userpost;
import com.zt.plat.module.databus.api.data.DatabusUserPostData;
/**
* 用户-岗位关系同步服务接口
* <p>
* 分公司需要实现此接口，完成数据的本地持久化
* 或通过默认实现 {@link UserPostSyncServiceImpl} 使用 Feign 调用远程 API
*
* @author ZT
*/
public interface UserPostSyncService {
/**
* 创建用户-岗位关系（增量同步）
*
* @param data 用户-岗位关系数据
*/
void create(DatabusUserPostData data);
/**
* 更新用户-岗位关系（增量同步）
*
* @param data 用户-岗位关系数据
*/
void update(DatabusUserPostData data);
/**
* 删除用户-岗位关系（增量同步）
*
* @param id 关系ID
*/
void delete(Long id);
/**
* 全量同步单条数据
* <p>
* 逻辑：存在则更新，不存在则插入
*
* @param data 用户-岗位关系数据
*/
void fullSync(DatabusUserPostData data);
}

56
zt-framework/zt-spring-boot-starter-databus-client/src/main/java/com/zt/plat/framework/databus/client/handler/userpost/UserPostSyncServiceImpl.java Normal file
View File

@@ -0,0 +1,56 @@
package com.zt.plat.framework.databus.client.handler.userpost;
import com.zt.plat.module.databus.api.data.DatabusUserPostData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
/**
* 用户-岗位关系同步服务实现
* <p>
* 使用条件：
* 1. zt.databus.sync.client.enabled=true
* <p>
* 注意：由于用户-岗位关系通常集成在用户管理中，此实现为占位符。
* 分公司可以根据实际情况：
* 1. 自定义实现此接口，直接操作本地数据库
* 2. 或者通过用户管理 API 间接处理关联关系
*
* @author ZT
*/
@Slf4j
@Service
@ConditionalOnProperty(prefix = "zt.databus.sync.client", name = "enabled", havingValue = "true")
public class UserPostSyncServiceImpl implements UserPostSyncService {
@Override
public void create(DatabusUserPostData data) {
log.info("[UserPostSync] 收到创建用户-岗位关系请求, userId={}, postId={}",
data.getUserId(), data.getPostId());
log.warn("[UserPostSync] 用户-岗位关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法，通过本地 API 或直接数据库操作完成同步
}
@Override
public void update(DatabusUserPostData data) {
log.info("[UserPostSync] 收到更新用户-岗位关系请求, userId={}, postId={}",
data.getUserId(), data.getPostId());
log.warn("[UserPostSync] 用户-岗位关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法
}
@Override
public void delete(Long id) {
log.info("[UserPostSync] 收到删除用户-岗位关系请求, id={}", id);
log.warn("[UserPostSync] 用户-岗位关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法
}
@Override
public void fullSync(DatabusUserPostData data) {
log.info("[UserPostSync] 收到全量同步用户-岗位关系请求, userId={}, postId={}",
data.getUserId(), data.getPostId());
log.warn("[UserPostSync] 用户-岗位关系同步服务需要分公司自定义实现，当前为占位符实现");
// TODO: 分公司需要实现此方法，逻辑：存在则更新，不存在则插入
}
}