突破数据孤岛：OceanBase CDC实现业务实时数据同步

2026-02-05 04:06:08作者：尤辰城Agatha

OceanBase is an enterprise distributed relational database with high availability, high performance, horizontal scalability, and compatibility with SQL standards.

项目地址：https://gitcode.com/GitHub_Trending/oc/oceanbase

你是否还在为数据库间数据同步延迟发愁？是否因无法实时捕获数据变更而错失业务良机？本文将详解OceanBase的CDC（Change Data Capture，数据变更捕获）功能，带你掌握从数据变更捕获到跨系统同步的完整解决方案，让你轻松实现业务数据的实时流动。

读完本文你将获得：

理解OceanBase CDC的核心架构与工作原理
掌握CDC功能的部署与配置方法
学会使用CDC进行数据同步的实战技巧
了解CDC在实际业务场景中的应用案例

什么是OceanBase CDC

CDC（Change Data Capture，数据变更捕获）是OceanBase提供的一项核心功能，它能够实时捕获数据库中的数据变更，并将这些变更以结构化的方式同步到其他系统。这一功能为企业构建实时数据管道、实现业务系统间的数据协同提供了强大支持。

OceanBase CDC功能主要通过src/logservice/cdcservice/和src/logservice/libobcdc/两个模块实现。其中，cdcservice模块负责数据变更的捕获和处理，而libobcdc模块则提供了客户端接口，方便用户消费变更数据。

OceanBase CDC的核心架构

OceanBase CDC功能采用了分层设计，主要包含以下几个核心组件：

CDC服务（ObCdcService）

CDC服务是整个功能的核心，负责协调各个组件的工作。它通过ObCdcService类实现，提供了数据变更捕获、日志读取、客户端连接管理等功能。

class ObCdcService: public lib::TGRunnable
{
public:
  int init(const uint64_t tenant_id, ObLSService *ls_service);
  void run1() override;
  int start();
  void stop();
  void wait();
  void destroy();
  
  // 核心接口
  int req_start_lsn_by_ts_ns(const obrpc::ObCdcReqStartLSNByTsReq &req,
      obrpc::ObCdcReqStartLSNByTsResp &resp);
  int fetch_log(const obrpc::ObCdcLSFetchLogReq &req,
      obrpc::ObCdcLSFetchLogResp &resp,
      const int64_t send_ts,
      const int64_t recv_ts);
  int fetch_raw_log(const obrpc::ObCdcFetchRawLogReq &req,
      obrpc::ObCdcFetchRawLogResp &resp,
      const int64_t send_ts,
      const int64_t recv_ts);
  // ...
};

日志获取器（ObCdcFetcher）

日志获取器负责从OceanBase的事务日志中读取变更数据。它支持两种获取模式：

在线模式（FETCHMODE_ONLINE）：直接从数据库实例的内存日志中读取变更
归档模式（FETCHMODE_ARCHIVE）：从归档存储中读取历史变更日志

这两种模式的切换由ClientLSCtx类管理，确保在不同场景下都能高效获取变更数据。

客户端上下文（ClientLSCtx）

客户端上下文用于管理每个CDC客户端连接的状态信息，包括：

客户端进度（client_progress_）
当前使用的协议类型（proto_type_）
日志获取模式（fetch_mode_）
流量统计信息（traffic_stat_info_）

这些信息通过ClientLSCtx类进行封装，确保每个客户端连接都能独立、高效地工作。

CDC数据同步流程

OceanBase CDC的数据同步流程主要包括以下几个步骤：

初始化CDC服务：通过ObCdcService::init()方法初始化CDC服务，设置租户ID、日志服务等关键参数。
定位起始LSN：客户端通过req_start_lsn_by_ts_ns()方法，根据时间戳定位到对应的日志序列号（LSN），作为数据同步的起点。
获取变更日志：客户端调用fetch_log()或fetch_raw_log()方法获取变更数据。其中：
- fetch_log()：获取格式化的日志条目
- fetch_raw_log()：获取原始日志数据
处理变更数据：客户端收到变更数据后，根据业务需求进行处理，如写入目标数据库、更新缓存等。
更新同步进度：客户端定期更新同步进度，确保在发生故障时能够从断点继续同步。

CDC工作流程图

sequenceDiagram
    participant Client
    participant ObCdcService
    participant ObCdcFetcher
    participant LogStorage
    
    Client->>ObCdcService: 初始化连接(init)
    Client->>ObCdcService: 请求起始LSN(req_start_lsn_by_ts_ns)
    ObCdcService->>LogStorage: 查询起始LSN
    LogStorage-->>ObCdcService: 返回起始LSN
    ObCdcService-->>Client: 返回起始LSN
    
    loop 数据同步循环
        Client->>ObCdcService: 获取变更日志(fetch_log)
        ObCdcService->>ObCdcFetcher: 读取日志
        ObCdcFetcher->>LogStorage: 读取日志数据
        LogStorage-->>ObCdcFetcher: 返回日志数据
        ObCdcFetcher-->>ObCdcService: 处理日志数据
        ObCdcService-->>Client: 返回变更数据
        Client->>Client: 处理变更数据
        Client->>ObCdcService: 更新同步进度
    end

部署与配置CDC功能

前提条件

在使用OceanBase CDC功能前，需要确保：

OceanBase集群已正常部署并运行
已启用归档日志（如需要同步历史数据）
具有足够权限的数据库用户

配置CDC服务

CDC服务的配置主要通过ObCdcService::init()方法完成：

int ObCdcService::init(const uint64_t tenant_id, ObLSService *ls_service)
{
  int ret = OB_SUCCESS;

  if (IS_INIT) {
    ret = OB_INIT_TWICE;;
    EXTLOG_LOG(WARN, "ObCdcService has inited", KR(ret));
  } else if (OB_FAIL(ls_ctx_map_.init("ExtClientLSCtxM", tenant_id))) {
    EXTLOG_LOG(WARN, "ls ctx map init failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(large_buffer_pool_.init("CDCService", 1L * 1024 * 1024 * 1024))) {
    EXTLOG_LOG(WARN, "large buffer pool init failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(log_ext_handler_.init())) {
    EXTLOG_LOG(WARN, "log ext handler init failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(locator_.init(tenant_id, this, &large_buffer_pool_, &log_ext_handler_))) {
    EXTLOG_LOG(WARN, "ObCdcStartLsnLocator init failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(fetcher_.init(tenant_id, ls_service, this, &large_buffer_pool_, &log_ext_handler_))) {
    EXTLOG_LOG(WARN, "ObCdcFetcher init failed", KR(ret), K(tenant_id));
  } else if (OB_FAIL(create_tenant_tg_(tenant_id))) {
    EXTLOG_LOG(WARN, "cdc thread group create failed", KR(ret), K(tenant_id));
  } else {
    dest_info_version_ = 0;
    tenant_id_ = tenant_id;
    is_inited_ = true;
  }

  return ret;
}

关键配置参数包括：

tenant_id：租户ID
large_buffer_pool_：缓冲区大小，默认为1GB
log_ext_handler_：日志外部存储处理器

启动CDC服务

CDC服务的启动通过ObCdcService::start()方法完成：

int ObCdcService::start()
{
  int ret = OB_SUCCESS;

  if (IS_NOT_INIT) {
    ret = OB_NOT_INIT;
    EXTLOG_LOG(WARN, "ObCdcService not init", K(ret));
  } else if (OB_FAIL(log_ext_handler_.start(0))) {
    EXTLOG_LOG(WARN, "log ext handler start failed", K(ret));
  } else if (OB_FAIL(start_tenant_tg_(tenant_id_))) {
    EXTLOG_LOG(ERROR, "start CDCService failed", KR(ret));
  } else {
    stop_flag_ = false;
  }

  return ret;
}

CDC功能的应用场景

实时数据仓库构建

利用CDC功能，可以将OceanBase中的业务数据实时同步到数据仓库，为企业决策提供实时数据支持。这种方式相比传统的批量ETL，具有以下优势：

同步延迟低，可实现准实时同步
对源数据库影响小，通过读取日志而非查询表
可捕获所有变更，包括插入、更新、删除操作

多活数据中心

在多活数据中心架构中，CDC功能可以用于实现数据中心之间的数据同步，确保各个数据中心的数据一致性。通过配置双向CDC同步，可以实现数据的双向流动，提高系统的可用性和容错能力。

微服务间数据同步

在微服务架构中，不同服务通常使用独立的数据库。CDC功能可以实现微服务间的数据同步，避免了传统的API调用方式带来的耦合和性能问题。

数据备份与恢复

CDC功能可以作为传统备份的补充，提供更细粒度的恢复能力。通过实时捕获数据变更，可以实现到任意时间点的恢复，大大减少数据丢失风险。

性能优化与最佳实践

调整缓冲区大小

OceanBase CDC使用large_buffer_pool_作为缓冲区，默认大小为1GB。在数据量较大的场景下，可以适当调大缓冲区大小：

// 默认初始化代码
if (OB_FAIL(large_buffer_pool_.init("CDCService", 1L * 1024 * 1024 * 1024))) {
  EXTLOG_LOG(WARN, "large buffer pool init failed", KR(ret), K(tenant_id));
}

合理设置同步频率

根据业务需求合理设置同步频率，在实时性和系统负载之间取得平衡。对于非关键业务，可以适当降低同步频率，减少系统资源消耗。

分库分表同步策略

对于分库分表场景，建议为每个表或表组单独配置CDC同步任务，避免单个同步任务负载过重。同时，可以通过ClientLSKey对不同表的同步上下文进行隔离：

ClientLSKey ls_key(client_id.get_addr(), client_id.get_pid(), client_tenant_id, ls_id);

监控与告警

OceanBase CDC提供了完善的监控指标，通过TrafficStatInfo类收集同步过程中的关键指标：

struct ClientLSTrafficStat
{
  ClientLSTrafficStat() { reset(); }
  ~ClientLSTrafficStat() { reset(); }
  void reset()
  {
    last_snapshot_time_ = ObTimeUtility::current_time();
    rpc_cnt_ = 0;
    time_stat_.reset();
  }
  
  int64_t last_snapshot_time_;
  int64_t rpc_cnt_;
  ClientLSTimeStat time_stat_;
};