企业信息化系统实时数据同步与双写架构设计

发布时间：2026-06-22 | 分类：架构设计

引言

在企业信息化系统的高可用架构设计中，跨机房数据同步与双写是保障业务连续性的关键技术。本文将探讨如何设计一套可靠的实时数据同步方案，确保在多活架构下数据的一致性与可用性。

双写架构概述

双写架构是指同时向多个数据源写入数据，确保当主数据源发生故障时，业务可以快速切换到备用数据源，保障服务的持续可用。

核心代码实现

双写代理核心实现：

// 双写管理器
class DualWriteManager {
  constructor(config) {
    this.sources = config.sources; // 数据源列表
    this.syncMode = config.syncMode || 'async'; // 同步模式
    this.conflictStrategy = config.conflictStrategy || 'last-write-wins';
    this.retryConfig = config.retry || { maxRetries: 3, backoff: 1000 };
    this.metrics = {
      totalWrites: 0,
      successfulWrites: 0,
      failedWrites: 0,
      conflictCount: 0
    };
  }

  // 写入操作
  async write(table, data, operation = 'insert') {
    this.metrics.totalWrites++;

    const writeTasks = this.sources.map(source =>
      this.writeToSource(source, table, data, operation)
    );

    try {
      // 并行写入所有数据源
      const results = await Promise.allSettled(writeTasks);

      // 分析写入结果
      const successResults = results.filter(r => r.status === 'fulfilled');
      const failedResults = results.filter(r => r.status === 'rejected');

      if (successResults.length === 0) {
        throw new Error('All sources write failed');
      }

      // 记录失败源用于后续补偿
      const failedSources = failedResults.map((r, i) => ({
        source: this.sources[i],
        error: r.reason
      }));

      if (failedSources.length > 0) {
        // 触发补偿任务
        await this.scheduleCompensation(failedSources, table, data, operation);
      }

      this.metrics.successfulWrites += successResults.length;
      this.metrics.failedWrites += failedSources.length;

      return {
        success: true,
        sourcesWritten: successResults.length,
        failedSources
      };

    } catch (error) {
      this.metrics.failedWrites++;
      throw error;
    }
  }

  // 写入单个数据源
  async writeToSource(source, table, data, operation) {
    const retryConfig = this.retryConfig;
    let lastError;

    for (let attempt = 0; attempt <= retryConfig.maxRetries; attempt++) {
      try {
        const connection = await this.getConnection(source);

        switch (operation) {
          case 'insert':
            await connection.insert(table, data);
            break;
          case 'update':
            await connection.update(table, data.where, data.values);
            break;
          case 'delete':
            await connection.delete(table, data.where);
            break;
        }

        await this.releaseConnection(source, connection);
        return { source: source.name, success: true };

      } catch (error) {
        lastError = error;

        if (attempt < retryConfig.maxRetries) {
          // 指数退避重试
          await this.sleep(retryConfig.backoff * Math.pow(2, attempt));
        }
      }
    }

    throw lastError;
  }

  // 批量写入优化
  async batchWrite(table, records, operation = 'insert') {
    const batchSize = 100;
    const batches = [];

    for (let i = 0; i < records.length; i += batchSize) {
      batches.push(records.slice(i, i + batchSize));
    }

    const results = [];
    for (const batch of batches) {
      const batchResult = await this.writeBatchToSources(table, batch, operation);
      results.push(batchResult);
    }

    return this.aggregateResults(results);
  }

  // 批量写入到所有数据源
  async writeBatchToSources(table, records, operation) {
    const tasks = this.sources.map(source =>
      this.writeBatchToSingleSource(source, table, records, operation)
    );

    const results = await Promise.allSettled(tasks);
    return results;
  }

  // 获取数据库连接
  async getConnection(source) {
    if (!source.pool) {
      source.pool = await createConnectionPool(source.config);
    }
    return await source.pool.acquire();
  }

  // 释放数据库连接
  async releaseConnection(source, connection) {
    await source.pool.release(connection);
  }

  // 延迟函数
  sleep(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }

  // 获取监控指标
  getMetrics() {
    return {
      ...this.metrics,
      successRate: this.metrics.totalWrites > 0
        ? (this.metrics.successfulWrites / this.metrics.totalWrites * 100).toFixed(2) + '%'
        : '0%'
    };
  }
}

// 数据同步服务
class SyncService {
  constructor(manager) {
    this.manager = manager;
    this.syncQueues = new Map();
    this.listeners = [];
  }

  // 监听数据变更
  async listenToChanges(sourceConfig) {
    const debezium = new DebeziumConnector({
      host: sourceConfig.host,
      port: sourceConfig.port,
      database: sourceConfig.database,
      tables: sourceConfig.tables,
      topic: sourceConfig.topic
    });

    await debezium.connect();

    debezium.on('change', async (change) => {
      await this.handleChange(change);
    });

    return debezium;
  }

  // 处理数据变更
  async handleChange(change) {
    const { operation, before, after, table, timestamp } = change;

    let op, data;
    switch (operation) {
      case 'c': // create
        op = 'insert';
        data = after;
        break;
      case 'u': // update
        op = 'update';
        data = { where: { id: before.id }, values: after };
        break;
      case 'd': // delete
        op = 'delete';
        data = { where: { id: before.id } };
        break;
      default:
        return;
    }

    // 执行双写
    try {
      await this.manager.write(table, data, op);

      // 通知监听器
      this.listeners.forEach(listener => {
        listener({ table, operation: op, data, timestamp });
      });
    } catch (error) {
      console.error('Sync failed:', error);
      // 写入重试队列
      await this.addToRetryQueue({ table, operation: op, data, timestamp });
    }
  }

  // 添加到重试队列
  async addToRetryQueue(item) {
    const queueKey = `${item.table}_${Date.now()}`;
    this.syncQueues.set(queueKey, {
      ...item,
      retryCount: 0,
      nextRetryTime: Date.now()
    });
  }

  // 重试队列处理
  async processRetryQueue() {
    const now = Date.now();
    const itemsToRetry = [];

    for (const [key, item] of this.syncQueues) {
      if (item.nextRetryTime <= now && item.retryCount < 5) {
        itemsToRetry.push({ key, item });
      }
    }

    for (const { key, item } of itemsToRetry) {
      try {
        await this.manager.write(item.table, item.data, item.operation);
        this.syncQueues.delete(key);
      } catch (error) {
        item.retryCount++;
        item.nextRetryTime = Date.now() + Math.pow(2, item.retryCount) * 1000;
      }
    }
  }
}

冲突解决策略

在双写场景下，如何处理并发写入冲突是关键问题：

// 冲突解决器
class ConflictResolver {
  constructor(strategy = 'last-write-wins') {
    this.strategy = strategy;
    this.customRules = new Map();
  }

  // 注册自定义冲突规则
  registerRule(table, handler) {
    this.customRules.set(table, handler);
  }

  // 解决冲突
  async resolve(source1, source2, table) {
    // 检查是否有自定义规则
    if (this.customRules.has(table)) {
      return await this.customRules.get(table)(source1, source2);
    }

    switch (this.strategy) {
      case 'last-write-wins':
        return this.lastWriteWins(source1, source2);

      case 'source-priority':
        return this.sourcePriority(source1, source2);

      case 'merge':
        return await this.merge(source1, source2);

      case 'manual':
        return this.manual(source1, source2);

      default:
        return this.lastWriteWins(source1, source2);
    }
  }

  // 最后写入胜出
  lastWriteWins(source1, source2) {
    const time1 = new Date(source1.metadata.updatedAt).getTime();
    const time2 = new Date(source2.metadata.updatedAt).getTime();
    return time1 > time2 ? source1 : source2;
  }

  // 源优先级
  sourcePriority(source1, source2) {
    const priority = { 'primary': 1, 'secondary': 2, 'tertiary': 3 };
    const p1 = priority[source1.metadata.source] || 99;
    const p2 = priority[source2.metadata.source] || 99;
    return p1 <= p2 ? source1 : source2;
  }

  // 合并策略
  async merge(source1, source2) {
    const merged = { ...source1 };

    for (const key of Object.keys(source2)) {
      // 字段级别比较时间戳
      if (source2.metadata && source2.metadata.fields[key]) {
        const time1 = source1.metadata.fields[key]?.updatedAt || 0;
        const time2 = source2.metadata.fields[key]?.updatedAt || 0;

        if (time2 > time1) {
          merged[key] = source2[key];
        }
      } else if (!(key in source1)) {
        merged[key] = source2[key];
      }
    }

    return merged;
  }

  // 标记需要人工处理
  manual(source1, source2) {
    return {
      conflict: true,
      sources: [source1, source2],
      requireManualResolution: true
    };
  }
}

// 版本向量用于冲突检测
class VersionVector {
  constructor() {
    this.versions = new Map();
  }

  // 更新版本
  update(source, timestamp) {
    const current = this.versions.get(source) || 0;
    this.versions.set(source, Math.max(current, timestamp));
  }

  // 检测冲突
  detectConflict(other) {
    const thisSources = new Set(this.versions.keys());
    const otherSources = new Set(other.versions.keys());

    // 检查共同源
    const commonSources = [...thisSources].filter(s => otherSources.has(s));

    for (const source of commonSources) {
      // 如果两个版本互相不知道对方的存在，则可能冲突
      if (!this.isDescendant(other, source) && !other.isDescendant(this, source)) {
        return true;
      }
    }

    return false;
  }

  isDescendant(other, source) {
    return this.versions.get(source) > other.versions.get(source);
  }
}

最佳实践建议

• 同步模式选择：根据业务需求选择同步或异步模式，对一致性要求高的场景采用同步模式
• 冲突检测：使用版本向量等技术提前检测可能的冲突，避免数据覆盖
• 补偿机制：建立完善的重试和补偿机制，处理临时性故障
• 监控告警：实时监控数据同步状态和延迟，设置合理的告警阈值
• 降级策略：当同步出现问题时，能够快速切换到单写模式保障可用性

总结

实时数据同步与双写架构是企业高可用系统的关键技术：

• 高可用：多数据源并行写入，单点故障不影响业务
• 一致性：通过版本向量和冲突解决机制保障数据一致性
• 可扩展：架构支持横向扩展，适应业务增长
• 可观测：完善的监控和告警体系，及时发现和处理问题

实际落地时需要根据业务场景和技术团队能力选择合适的方案，并持续优化迭代。