如何实现Flutter Hive本地库与PostgreSQL后端库的自动数据同步（CDC）

一、核心流程设计 #

1. 本地端变更捕获（模拟CDC） #

Hive作为本地NoSQL存储无内置CDC机制，需手动实现变更日志记录：

• 给所有Hive实体添加元数据字段：lastModifiedAt（时间戳）、syncStatus（pending/synced/failed）
• 单独维护一个Hive盒子（Box）存储变更日志，每条日志包含：操作类型（create/update/delete）、实体类型、数据ID、数据内容、时间戳、同步状态
• 在仓储层封装增删改操作，确保数据写入与日志记录原子性（用Hive的boxTransaction事务）

2. 联网触发同步 #

• 用connectivity_plus监听网络状态，从离线切换到在线时自动触发同步
• 应用启动、后台唤醒时主动检查未同步数据，触发同步流程

3. 本地数据上传至后端 #

• 从变更日志中筛选syncStatus=pending的记录，按时间戳排序（保证操作顺序）
• 调用后端批量API提交变更，后端按操作类型处理：
  • Create：插入PostgreSQL，处理唯一键冲突（如本地重复创建）
  • Update：用lastModifiedAt做乐观锁，仅当本地时间戳晚于后端时更新，避免覆盖最新数据
  • Delete：根据业务选择软删除（标记is_deleted=true）或硬删除

4. 后端数据拉取至本地 #

• 上传完成后，传递本地最新同步时间lastSyncedAt，请求后端返回该时间后的增量变更
• 本地接收后更新Hive数据，标记对应变更日志为synced，更新lastSyncedAt

二、PostgreSQL端CDC落地实现 #

PostgreSQL有两种主流CDC方案适配移动端同步需求：

方案1：触发器+变更表（轻量易用） #

创建专门的变更记录表，通过触发器捕获所有增删改操作：

-- 创建变更记录表
CREATE TABLE entity_changelog (
    id SERIAL PRIMARY KEY,
    entity_type VARCHAR(50) NOT NULL,
    entity_id VARCHAR(100) NOT NULL,
    operation_type VARCHAR(10) NOT NULL,
    data JSONB NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 为product表创建变更触发器
CREATE OR REPLACE FUNCTION log_product_change()
RETURNS TRIGGER AS $$
BEGIN
    IF TG_OP = 'INSERT' THEN
        INSERT INTO entity_changelog VALUES ('product', NEW.id, 'create', row_to_json(NEW), CURRENT_TIMESTAMP);
    ELSIF TG_OP = 'UPDATE' THEN
        INSERT INTO entity_changelog VALUES ('product', NEW.id, 'update', row_to_json(NEW), CURRENT_TIMESTAMP);
    ELSIF TG_OP = 'DELETE' THEN
        INSERT INTO entity_changelog VALUES ('product', OLD.id, 'delete', row_to_json(OLD), CURRENT_TIMESTAMP);
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER product_change_trigger
AFTER INSERT OR UPDATE OR DELETE ON product
FOR EACH ROW EXECUTE FUNCTION log_product_change();

后端提供增量查询API，根据客户端传入的lastSyncedAt返回对应时间段的变更记录。

方案2：WAL解析（适合大规模场景） #

使用wal2json或pg_logical插件解析PostgreSQL的WAL日志，直接获取变更流。这种方案适合实时推送场景，但移动端通常用拉取模式，触发器方案更轻量。

三、最佳实践 #

1. 冲突处理与乐观锁 #

• 本地与后端均维护lastModifiedAt，同步时对比时间戳：若后端数据更新时间更晚，提示用户手动合并或按业务规则自动处理
• 用version版本号替代时间戳，更新时检查版本一致性，不一致则拒绝更新，避免并发冲突

2. 可靠同步机制 #

• 批量提交变更记录，减少API调用次数
• 同步失败的记录标记为failed，下次联网自动重试，重试3次后提示用户手动干预
• 本地操作必须原子化：数据写入与变更日志记录放在同一Hive事务中

3. 数据安全 #

• Hive数据加密（使用hive_flutter的加密功能）
• 传输过程强制HTTPS
• PostgreSQL启用数据加密与访问控制

4. 用户体验优化 #

• 在应用内展示同步状态（如“正在同步”“同步完成”“同步失败”）
• 离线操作时给出明确反馈，避免用户疑惑
• 大体积数据同步时显示进度条

5. 避免重复同步 #

• 每次同步完成后更新本地lastSyncedAt，仅同步该时间点后的变更
• 后端记录已同步的客户端ID与变更ID，防止重复处理

四、代码示例（Flutter端） #

变更日志实体 #

import 'package:hive/hive.dart';

part 'change_log.g.dart';

@HiveType(typeId: 1)
class ChangeLog {
  @HiveField(0)
  final String id;

  @HiveField(1)
  final String entityType;

  @HiveField(2)
  final String entityId;

  @HiveField(3)
  final String operationType;

  @HiveField(4)
  final Map<String, dynamic> data;

  @HiveField(5)
  final DateTime timestamp;

  @HiveField(6)
  String syncStatus;

  ChangeLog({
    required this.id,
    required this.entityType,
    required this.entityId,
    required this.operationType,
    required this.data,
    required this.timestamp,
    this.syncStatus = "pending",
  });
}

仓储层原子操作 #

class ProductRepository {
  final Box<Product> _productBox;
  final Box<ChangeLog> _changeLogBox;

  ProductRepository(this._productBox, this._changeLogBox);

  Future<void> addProduct(Product product) async {
    await Hive.boxTransaction([_productBox, _changeLogBox], () async {
      await _productBox.put(product.id, product);
      await _changeLogBox.put(
        DateTime.now().millisecondsSinceEpoch.toString(),
        ChangeLog(
          id: DateTime.now().millisecondsSinceEpoch.toString(),
          entityType: "product",
          entityId: product.id,
          operationType: "create",
          data: product.toJson(),
          timestamp: DateTime.now(),
        ),
      );
    });
  }
}

内容的提问来源于stack exchange，提问作者reza te