fix(databus): 修复客户端消息处理和防止消息循环
1. 修复消息格式不匹配问题 - 增量消息:兼容 SyncMessage 格式,从 dataSnapshot 字段反序列化数据 - 批量消息:添加 getDataType() 方法获取泛型类型,正确转换 JSONObject 2. 防止消息循环 - 添加 zt.databus.change.producer.enabled 配置项 - 客户端禁用变更消息发送,避免 客户端写入 → 发送变更 → 循环 3. 修复 Feign 客户端注入 - 在 RpcConfiguration 中添加 DeptApi、PostApi - 确保客户端能通过 Feign 调用本地 system-server API 相关文件: - DatabusClientConsumer.java: 修复消息解析逻辑 - BatchSyncEventHandler.java: 添加 getDataType() 方法 - DatabusChangeProducer.java: 添加 enabled 开关 - RpcConfiguration.java: 启用 DeptApi/PostApi Feign 客户端 Ref: 修复 ClassCastException 和消息循环问题
This commit is contained in:
@@ -56,7 +56,7 @@ public class DatabusClientConsumer implements RocketMQListener<String> {
|
||||
log.info("[DatabusClient] 收到消息, eventType={}", eventType);
|
||||
|
||||
// 2. 根据 eventType 判断消息类型并分发
|
||||
if (isBatchMessage(body)) {
|
||||
if (isBatchMessage(eventType)) {
|
||||
// 批量消息(全量同步)
|
||||
handleBatchMessage(body, eventType);
|
||||
} else {
|
||||
@@ -82,18 +82,60 @@ public class DatabusClientConsumer implements RocketMQListener<String> {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 解析批量消息
|
||||
DatabusBatchMessage<?> message = JSON.parseObject(body, DatabusBatchMessage.class);
|
||||
// 2. 获取数据类型
|
||||
Class<?> dataType = handler.getDataType();
|
||||
|
||||
// 3. 全量同步开始回调(第一批)
|
||||
// 3. 解析批量消息(兼容服务端 BatchSyncMessage 格式)
|
||||
var json = JSON.parseObject(body);
|
||||
|
||||
// 兼容处理:服务端使用 fullTaskId,转换为 taskId
|
||||
String taskId = json.getString("taskId");
|
||||
if (taskId == null) {
|
||||
taskId = String.valueOf(json.getLong("fullTaskId"));
|
||||
}
|
||||
|
||||
DatabusBatchMessage<Object> message = new DatabusBatchMessage<>();
|
||||
message.setMessageId(json.getString("messageId"));
|
||||
message.setTaskId(taskId);
|
||||
message.setEventType(eventType);
|
||||
message.setBatchNo(json.getInteger("batchNo"));
|
||||
message.setTotalBatch(json.getInteger("totalBatch"));
|
||||
message.setCount(json.getInteger("count"));
|
||||
message.setTotalCount(json.getInteger("totalCount") != null ? json.getInteger("totalCount") : 0);
|
||||
message.setIsLastBatch(json.getBoolean("isLastBatch"));
|
||||
message.setTenantId(json.getLong("tenantId"));
|
||||
|
||||
// 解析 dataList(服务端是 SyncDataItem 列表,需要提取 data 字段并转换为具体类型)
|
||||
var dataListJson = json.getJSONArray("dataList");
|
||||
if (dataListJson != null) {
|
||||
java.util.List<Object> dataList = new java.util.ArrayList<>();
|
||||
for (int i = 0; i < dataListJson.size(); i++) {
|
||||
var item = dataListJson.getJSONObject(i);
|
||||
// 服务端 SyncDataItem 结构:{action, uid, data}
|
||||
// data 字段是 JSON 字符串,需要解析成具体类型
|
||||
String dataStr = item.getString("data");
|
||||
if (dataStr != null) {
|
||||
// 使用 handler.getDataType() 反序列化成具体类型
|
||||
Object data = JSON.parseObject(dataStr, dataType);
|
||||
dataList.add(data);
|
||||
} else {
|
||||
// 如果没有 data 字段,可能是直接的数据对象,也需要转换类型
|
||||
Object data = JSON.parseObject(item.toJSONString(), dataType);
|
||||
dataList.add(data);
|
||||
}
|
||||
}
|
||||
message.setDataList(dataList);
|
||||
}
|
||||
|
||||
// 4. 全量同步开始回调(第一批)
|
||||
if (message.getBatchNo() == 1) {
|
||||
handler.onFullSyncStart(message);
|
||||
}
|
||||
|
||||
// 4. 处理批次数据
|
||||
// 5. 处理批次数据
|
||||
handler.handleBatch(message);
|
||||
|
||||
// 5. 全量同步完成回调(最后一批)
|
||||
// 6. 全量同步完成回调(最后一批)
|
||||
if (message.getBatchNo().equals(message.getTotalBatch())) {
|
||||
handler.onFullSyncComplete(message);
|
||||
}
|
||||
@@ -104,6 +146,15 @@ public class DatabusClientConsumer implements RocketMQListener<String> {
|
||||
|
||||
/**
|
||||
* 处理增量消息
|
||||
* <p>
|
||||
* 兼容服务端 SyncMessage 格式:
|
||||
* - syncId: 同步ID
|
||||
* - eventRecordId: 事件记录ID
|
||||
* - eventType: 事件类型
|
||||
* - eventAction: 事件动作
|
||||
* - dataSnapshot: 业务数据快照(JSON字符串)
|
||||
* - dataVersion: 数据版本
|
||||
* - timestamp: 时间戳
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private void handleIncrementalMessage(String body, DatabusEventType eventType) {
|
||||
@@ -114,14 +165,36 @@ public class DatabusClientConsumer implements RocketMQListener<String> {
|
||||
return;
|
||||
}
|
||||
|
||||
// 2. 解析增量消息
|
||||
DatabusMessage<?> message = JSON.parseObject(body, DatabusMessage.class);
|
||||
// 2. 解析增量消息(兼容服务端 SyncMessage 格式)
|
||||
var json = JSON.parseObject(body);
|
||||
|
||||
// 从 dataSnapshot 字段解析业务数据
|
||||
String dataSnapshot = json.getString("dataSnapshot");
|
||||
Object data = null;
|
||||
Long dataId = null;
|
||||
if (dataSnapshot != null && !dataSnapshot.isEmpty()) {
|
||||
// dataSnapshot 是 JSON 字符串,需要解析成具体类型
|
||||
var dataJson = JSON.parseObject(dataSnapshot);
|
||||
// 获取数据类型并反序列化
|
||||
Class<?> dataType = handler.getDataType();
|
||||
data = JSON.parseObject(dataSnapshot, dataType);
|
||||
// 提取 dataId
|
||||
dataId = dataJson.getLong("id");
|
||||
}
|
||||
|
||||
// 构建 DatabusMessage
|
||||
DatabusMessage<Object> message = new DatabusMessage<>();
|
||||
message.setMessageId(json.getString("syncId"));
|
||||
message.setEventType(eventType);
|
||||
message.setDataId(dataId);
|
||||
message.setData(data);
|
||||
message.setTenantId(json.getLong("tenantId"));
|
||||
|
||||
// 3. 处理消息
|
||||
handler.handle(message);
|
||||
|
||||
log.info("[DatabusClient] 增量消息处理完成, eventType={}, messageId={}",
|
||||
eventType, message.getMessageId());
|
||||
log.info("[DatabusClient] 增量消息处理完成, eventType={}, syncId={}, dataId={}",
|
||||
eventType, message.getMessageId(), dataId);
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -143,13 +216,11 @@ public class DatabusClientConsumer implements RocketMQListener<String> {
|
||||
/**
|
||||
* 判断是否为批量消息
|
||||
*/
|
||||
private boolean isBatchMessage(String body) {
|
||||
private boolean isBatchMessage(DatabusEventType eventType) {
|
||||
try {
|
||||
// 批量消息包含 taskId, batchNo, totalBatch 字段
|
||||
var json = JSON.parseObject(body);
|
||||
return json.containsKey("taskId")
|
||||
&& json.containsKey("batchNo")
|
||||
&& json.containsKey("totalBatch");
|
||||
// 批量消息包含 batchNo, totalBatch 字段
|
||||
// 服务端使用 fullTaskId,客户端 API 使用 taskId,兼容两种格式
|
||||
return eventType.getAction().equals("full");
|
||||
} catch (Exception e) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@@ -3,6 +3,9 @@ package com.zt.plat.framework.databus.client.handler;
|
||||
import com.zt.plat.module.databus.api.message.DatabusBatchMessage;
|
||||
import com.zt.plat.module.databus.enums.DatabusEventType;
|
||||
|
||||
import java.lang.reflect.ParameterizedType;
|
||||
import java.lang.reflect.Type;
|
||||
|
||||
/**
|
||||
* 批量同步事件处理器接口
|
||||
* <p>
|
||||
@@ -48,4 +51,29 @@ public interface BatchSyncEventHandler<T> {
|
||||
// 默认空实现,子类可覆盖
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取数据类型
|
||||
* <p>
|
||||
* 默认通过反射获取泛型类型参数,子类可以覆盖此方法提供具体类型
|
||||
*
|
||||
* @return 数据类型的 Class 对象
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
default Class<T> getDataType() {
|
||||
Type[] genericInterfaces = this.getClass().getGenericInterfaces();
|
||||
for (Type genericInterface : genericInterfaces) {
|
||||
if (genericInterface instanceof ParameterizedType) {
|
||||
ParameterizedType parameterizedType = (ParameterizedType) genericInterface;
|
||||
if (parameterizedType.getRawType().equals(BatchSyncEventHandler.class)) {
|
||||
Type[] typeArguments = parameterizedType.getActualTypeArguments();
|
||||
if (typeArguments.length > 0 && typeArguments[0] instanceof Class) {
|
||||
return (Class<T>) typeArguments[0];
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// 如果无法获取泛型类型,返回 Object.class
|
||||
return (Class<T>) Object.class;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user