SQL数据同步是数据库运维和数据处理领域的常见需求，当业务需要将源数据库的数据同步到目标数据库，或者多个数据库之间需要保持数据一致时，就需要选择合适的同步方案。增量同步和实时同步是两种最常用的同步模式，适用场景和实现逻辑存在明显差异。
SQL增量同步方案
增量同步指的是只同步源数据库中发生变化的数据，避免全量同步带来的资源浪费，适合数据量较大、变更频率较低的场景。
核心实现原理
增量同步的核心是识别源库中新增、修改、删除的数据，常见的识别方式有三种：
- 基于时间戳字段：源表新增
update_time字段，每次数据变更时更新该字段，同步时查询大于上次同步时间的数据 - 基于自增ID：源表使用自增主键，同步时记录上次同步的最大ID，下次同步查询ID大于该值的数据
- 基于日志解析：解析数据库的binlog（MySQL）、WAL（PostgreSQL）等日志文件，获取所有数据变更记录
基于时间戳的增量同步示例
假设源表为user_info，包含id、username、update_time字段，目标表结构相同，以下是使用Python实现的增量同步逻辑：
import pymysql
import time
# 连接源数据库
source_conn = pymysql.connect(
host='127.0.0.1',
user='root',
password='123456',
database='source_db',
charset='utf8mb4'
)
# 连接目标数据库
target_conn = pymysql.connect(
host='127.0.0.1',
user='root',
password='123456',
database='target_db',
charset='utf8mb4'
)
# 记录上次同步时间，实际场景可存储到文件或配置表
last_sync_time = '2024-01-01 00:00:00'
def incremental_sync():
source_cursor = source_conn.cursor()
target_cursor = target_conn.cursor()
# 查询源库中变更的数据
query_sql = "SELECT id, username, update_time FROM user_info WHERE update_time > %s"
source_cursor.execute(query_sql, (last_sync_time,))
changed_data = source_cursor.fetchall()
for data in changed_data:
user_id, username, update_time = data
# 判断目标库是否存在该数据，存在则更新，不存在则插入
check_sql = "SELECT id FROM user_info WHERE id = %s"
target_cursor.execute(check_sql, (user_id,))
if target_cursor.fetchone():
update_sql = "UPDATE user_info SET username=%s, update_time=%s WHERE id=%s"
target_cursor.execute(update_sql, (username, update_time, user_id))
else:
insert_sql = "INSERT INTO user_info (id, username, update_time) VALUES (%s, %s, %s)"
target_cursor.execute(insert_sql, (user_id, username, update_time))
target_conn.commit()
# 更新上次同步时间
new_sync_time = time.strftime('%Y-%m-%d %H:%M:%S')
print(f"同步完成，本次同步时间：{new_sync_time}")
if __name__ == '__main__':
incremental_sync()
source_conn.close()
target_conn.close()
SQL实时同步方案
实时同步要求源库数据发生变更后，目标库在极短时间内完成同步，适合对数据一致性要求高、变更频率高的场景，比如业务库到实时数仓的同步。
核心实现原理
实时同步通常基于数据库的变更日志实现，通过监听日志的变更事件，实时触发同步操作，常见的实现方式有两种：
- 基于开源工具：使用Canal、Debezium等工具监听数据库日志，将变更事件发送到消息队列，再由消费程序同步到目标库
- 基于数据库自身能力：部分数据库支持原生的同步功能，比如MySQL的主从复制、PostgreSQL的逻辑复制，可直接实现实时同步
基于Canal的实时同步示例
Canal是阿里巴巴开源的MySQL binlog解析工具，可模拟MySQL从库交互协议，获取binlog变更记录。以下是简单的配置和同步逻辑示例：
首先修改MySQL配置文件my.cnf，开启binlog：
[mysqld] log-bin=mysql-bin binlog-format=ROW server-id=1
Canal的配置文件instance.properties核心配置：
# 源库地址 canal.instance.master.address=127.0.0.1:3306 # 源库账号密码 canal.instance.dbUsername=root canal.instance.dbPassword=123456 # 监听的数据库和表 canal.instance.filter.regex=source_db\.user_info
Java消费Canal变更事件的示例代码：
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import java.net.InetSocketAddress;
import java.util.List;
public class CanalSyncDemo {
public static void main(String[] args) {
// 创建链接
CanalConnector connector = CanalConnectors.newSingleConnector(
new InetSocketAddress("127.0.0.1", 11111),
"example",
"",
""
);
connector.connect();
connector.subscribe("source_db\.user_info");
connector.rollback();
while (true) {
// 获取指定数量的数据
Message message = connector.getWithoutAck(100);
long batchId = message.getId();
List<Entry> entries = message.getEntries();
if (batchId != -1 && entries.size() > 0) {
for (Entry entry : entries) {
// 解析binlog变更内容
RowChange rowChange = RowChange.parseFrom(entry.getStoreValue());
// 处理新增、修改、删除事件，同步到目标库
System.out.println("变更类型：" + rowChange.getEventType());
// 此处添加同步到目标库的逻辑
}
}
connector.ack(batchId);
}
}
}
两种方案对比
增量同步和实时同步的特性差异如下，可根据业务需求选择：
| 对比维度 | 增量同步 | 实时同步 |
|---|---|---|
| 同步延迟 | 分钟级或小时级 | 秒级或毫秒级 |
| 资源消耗 | 较低，仅同步变更数据 | 较高，需要持续监听日志 |
| 实现复杂度 | 较低，可自定义脚本实现 | 较高，依赖中间件或数据库原生能力 |
| 适用场景 | 数据量较大、变更频率低的离线同步场景 | 对数据一致性要求高的实时业务场景 |
方案选择建议
如果业务对同步延迟要求不高，数据变更频率较低，优先选择增量同步，实现成本低且运维简单。如果业务需要数据实时一致，比如实时报表、实时风控场景，优先选择实时同步方案。另外如果源库是MySQL，数据量较大且需要实时同步，可优先使用Canal+消息队列的组合方案，扩展性和稳定性更好。