本文目录导读：
- 基于消息队列（Kafka/RocketMQ）的实时同步
- 基于Redis Pub/Sub的实时同步
- 基于CDC（Change Data Capture）的实时同步
- 基于gRPC双向流的实时同步
- 组合方案（推荐）
- 关键考虑因素
在Java中实现分布式数据实时同步API，可以结合多种技术方案,以下是几种主流且成熟的实现方式：
基于消息队列（Kafka/RocketMQ）的实时同步
核心架构
Source App → 消息队列 → Target App示例代码（使用Kafka）
// 生产者端（数据源）
@Component
public class DataSyncProducer {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void syncData(String operation, Object data) {
SyncMessage message = new SyncMessage();
message.setOperation(operation); // INSERT/UPDATE/DELETE
message.setData(JSON.toJSONString(data));
message.setTimestamp(System.currentTimeMillis());
// 发送到同步主题
kafkaTemplate.send("data-sync-topic",
message.getDataId(),
JSON.toJSONString(message));
}
}
// 消费者端（目标应用）
@Component
public class DataSyncConsumer {
@KafkaListener(topics = "data-sync-topic",
groupId = "data-sync-group")
public void handleSync(String message) {
SyncMessage syncMsg = JSON.parseObject(message, SyncMessage.class);
switch(syncMsg.getOperation()) {
case "INSERT":
// 插入数据到本地缓存/数据库
cacheManager.put(syncMsg.getDataId(), syncMsg.getData());
break;
case "UPDATE":
cacheManager.update(syncMsg.getDataId(), syncMsg.getData());
break;
case "DELETE":
cacheManager.remove(syncMsg.getDataId());
break;
}
}
}
基于Redis Pub/Sub的实时同步
适用于轻量级场景
@Component
public class RedisDataSync {
@Autowired
private StringRedisTemplate redisTemplate;
// 发布数据变更
public void publishChange(String channel, DataChangeEvent event) {
redisTemplate.convertAndSend(channel,
JSON.toJSONString(event));
}
}
// 订阅端配置
@Configuration
public class RedisSubscriberConfig {
@Bean
public RedisMessageListenerContainer container(
RedisConnectionFactory factory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container =
new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
container.addMessageListener(
listenerAdapter,
new PatternTopic("data-sync:*"));
return container;
}
@Bean
public MessageListenerAdapter listenerAdapter() {
return new MessageListenerAdapter(new DataSyncListener(),
"handleMessage");
}
}
// 监听器实现
@Component
public class DataSyncListener {
public void handleMessage(String message) {
DataChangeEvent event = JSON.parseObject(message,
DataChangeEvent.class);
// 处理数据同步
processEvent(event);
}
}
基于CDC（Change Data Capture）的实时同步
使用Debezium + Kafka Connect
// Debezium配置自动监听数据库变更
@Configuration
public class DebeziumConfig {
@Bean
public io.debezium.config.Configuration debeziumConfig() {
return io.debezium.config.Configuration.create()
.with("connector.class",
"io.debezium.connector.mysql.MySqlConnector")
.with("database.hostname", "localhost")
.with("database.port", 3306)
.with("database.user", "debezium")
.with("database.password", "dbz")
.with("database.server.id", "184054")
.with("database.server.name", "source-db")
.with("database.include.list", "mydb")
.with("table.include.list", "mydb.users")
.with("database.history",
"io.debezium.relational.history.MemoryDatabaseHistory")
.build();
}
}
// 监听数据变更事件
@Component
public class DebeziumChangeListener {
@EventListener
public void handleChangeEvent(ChangeEvent<String, String> event) {
String value = event.value();
JsonNode changeNode = Json.parse(value);
// 解析变更操作
String op = changeNode.get("payload").get("op").asText();
JsonNode after = changeNode.get("payload").get("after");
switch(op) {
case "c": // Create
handleCreate(after);
break;
case "u": // Update
handleUpdate(after);
break;
case "d": // Delete
handleDelete(changeNode.get("payload").get("before"));
break;
}
// 同步到目标系统
syncToTarget(changeNode);
}
}
基于gRPC双向流的实时同步
支持高并发、低延迟
// proto定义
service DataSyncService {
rpc SyncData(stream SyncRequest) returns (stream SyncResponse);
}
message SyncRequest {
string operation = 1;
string data_id = 2;
bytes data = 3;
int64 timestamp = 4;
}
message SyncResponse {
bool success = 1;
string message = 2;
int64 server_time = 3;
}
// 服务器端实现
@GrpcService
public class DataSyncServiceImpl extends DataSyncServiceGrpc.DataSyncServiceImplBase {
@Override
public StreamObserver<SyncRequest> syncData(
StreamObserver<SyncResponse> responseObserver) {
return new StreamObserver<SyncRequest>() {
@Override
public void onNext(SyncRequest request) {
// 处理数据同步
boolean success = processSyncRequest(request);
SyncResponse response = SyncResponse.newBuilder()
.setSuccess(true)
.setMessage("synced")
.setServerTime(System.currentTimeMillis())
.build();
responseObserver.onNext(response);
}
@Override
public void onError(Throwable t) {
log.error("Sync error", t);
}
@Override
public void onCompleted() {
responseObserver.onCompleted();
}
};
}
}
组合方案（推荐）
结合多种技术实现最佳效果
// 统一的数据同步管理器
@Component
public class DataSyncManager {
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private ApplicationEventPublisher eventPublisher;
public void syncData(String tableName, String operation, Object data) {
// 1. 发布到Kafka（持久化）
kafkaTemplate.send("data-sync-" + tableName,
generateKey(data),
serializeData(operation, data));
// 2. 发布到Redis Pub/Sub（实时通知）
redisTemplate.convertAndSend(
"data-change:" + tableName,
serializeData(operation, data)
);
// 3. 触发本地事件
eventPublisher.publishEvent(new DataChangeEvent(
tableName, operation, data
));
}
// 异步处理同步逻辑
@Async
@EventListener
public void handleAsyncSync(DataChangeEvent event) {
// 1. 更新本地缓存
updateLocalCache(event);
// 2. 同步到其他服务
syncToOtherServices(event);
// 3. 更新搜索引擎索引
updateSearchIndex(event);
}
}
关键考虑因素
- 数据一致性：使用分布式事务（Seata/TCC）或最终一致性方案
- 性能优化：批量处理、异步写入、连接池管理
- 容错处理：重试机制、死信队列、幂等性设计
- 监控告警：集成Prometheus/Grafana监控同步延迟和成功率
选择合适的方案时，需要根据业务需求（延迟要求、数据量、一致性级别等）和技术栈进行权衡。
