跨EF Core上下文同步大型数据库表的高效实现方案

云上老周

2026-8-27

我的场景

我有3个编号为1、2、3的仓库数据库（Firebird），所有库采用相同架构、共用同一个DbContext类，Products表的模型定义如下：

public class Product
{
    public string Sku { get; }
    public string Barcode { get; }
    public int Quantity { get; }
}

我另有一个本地“仓库缓存”数据库（MySQL），需要定期拉取三个仓库的全量数据用于缓存。缓存产品的数据模型与源端类似，额外新增了标识来源仓库编号的字段，该表需要存储三个仓库的全部产品信息：若同Sku产品同时出现在1号和3号仓库，缓存表中需要保留两条独立记录，分别对应所属的仓库ID，CachedProduct模型定义如下：

public class CachedProduct
{
    public int WarehouseId { get; set; } // 取值为1、2、3
    public string Sku { get; }
    public string Barcode { get; }
    public int Quantity { get; }
}

单仓库数据量约为2万条，我尝试过多种方案均存在可行性或效率问题，希望获取更优的实现思路。

核心问题

本地缓存库为空时实现非常简单，直接拉取三个仓库的全量产品数据写入缓存库即可。但后续同步时缓存库已有存量数据，此时不能重复插入全量6万条数据浪费存储空间，需要实现upsert逻辑：新到的产品数据正常insert，若缓存中已存在匹配记录（匹配条件为Sku+WarehouseId），则直接update对应记录（例如上次同步后某仓库的Quantity库存数量发生变动），最终保证缓存库的记录数始终等于三个仓库的记录总和，无冗余也无缺失。

已尝试的方案

逐行处理的贪心方案：实现逻辑最简单，遍历每个仓库的每一条产品数据，检查缓存表中是否存在匹配记录，存在则执行update，不存在则执行insert。该方案的明显缺陷是无法做批量优化，每次同步会产生数万次select、insert、update数据库调用，效率极低。
全量清空重导方案：每次同步前先清空本地缓存库，再重新拉取全量数据写入。该方案的问题是同步过程中存在短暂的缓存空窗期，会导致应用其他模块无法读取缓存数据，引发业务异常。
使用EF Core生态Upsert库方案：我最初尝试了支持批量操作的FlexLabs.Upsert库，原本预期效果最好，但实测该库存在异常，即使运行官方最小示例也无法正常工作：无论匹配规则如何配置，每次执行“upsert”都会插入新行，完全不触发更新逻辑。
弃用EF Core使用专用同步库方案：我调研了数据库间专用同步库Dotmim.Sync，但该库不支持源端的FirebirdDB，同时我不确定该库是否支持同步过程中的数据转换——写入缓存库前需要为每行数据添加对应的WarehouseId字段值。

实现方案

6万条总数据量属于非常小的同步规模，不需要引入重型同步组件，用「临时表+数据库原生批量upsert」的方案就能兼顾性能和可用性，全程无空窗期，实现逻辑也不复杂：

先给MySQL的CachedProduct表创建联合唯一索引，这是所有upsert逻辑的基础，同时能保证匹配查询的性能：
```
CREATE UNIQUE INDEX idx_warehouse_sku ON CachedProducts(WarehouseId, Sku);
```
之前用FlexLabs.Upsert不触发更新、只插入新行，大概率就是没建这个唯一索引，库生成的upsert语句找不到唯一约束判断匹配条件，自然只会走插入逻辑。
同步时按仓库拉取全量Product数据，在内存中直接组装成带对应WarehouseId的CachedProduct集合，6万条数据的内存占用不到10MB，完全没有压力。
走批量写入流程，全程只有3次数据库交互，秒级完成同步：
- 第一步：在当前数据库连接创建会话级临时表，表结构和CachedProduct完全一致，临时表仅当前连接可见，不会影响线上业务读缓存。
- 第二步：用EF Core自带的批量写入能力（EF Core 7.0及以上版本原生支持批量插入，低版本可以用成熟的EF批量扩展包），把组装好的6万条CachedProduct一次性写入临时表，批量写入的单批次提交性能是逐行写入的上百倍。
- 第三步：执行单条MySQL原生upsert语句，把临时表的数据同步到正式缓存表，命中唯一索引（WarehouseId+Sku已存在）就更新Barcode、Quantity字段，未命中就直接插入新行：
```
INSERT INTO CachedProducts (WarehouseId, Sku, Barcode, Quantity)
SELECT WarehouseId, Sku, Barcode, Quantity FROM temp_cache_products
ON DUPLICATE KEY UPDATE
    Barcode = VALUES(Barcode),
    Quantity = VALUES(Quantity);
```
如果需要同步源端的删除操作（比如某个仓库下架了Sku，缓存里也要删掉对应记录），只需要在upsert完成后加一条删除语句，删掉正式表中对应WarehouseId、但Sku不在临时表中的记录即可，执行完销毁临时表，同步完成。

这个方案的优势很明显：

没有逐行查询、逐行写入的开销，6万条数据同步全程耗时基本在1秒以内。
全程操作正式表时走MySQL行级锁，不会锁全表，也没有缓存空窗期，业务侧读缓存完全不受影响。
不依赖有兼容性问题的第三方Upsert库，核心逻辑用数据库原生语法实现，稳定可控。
后续数据量上涨到几十万条级别，这个方案的性能依然足够，不需要调整架构。

内容的提问来源于stack exchange，提问作者Lázár Zsolt