--- title: "obcdc 开发说明 - OceanBase 数据库 V4.6.0 | OceanBase 文档中心" description: obcdc 开发说明 本文介绍如何将社区版 obcdc 接入自己的数据消费链路。 功能适用性 该内容仅适用于社区版 obcdc。 使用 obcdc 动态库开发自己的 OceanBase 数据消费工具 obcdc 是 C++ 编写，编译产出物为动态库，下游消费程序开发时需要依赖动态库和头文件（libobcdc.h、ob… --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 4.6.0 # obcdc 开发说明 更新时间：2026-06-19 10:13:53 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V4.6.0/zh-CN/700.reference/1500.Components-and-Tools/300.data-integrate/400.cdc/250.obcdc-development-instructions.md) 本文介绍如何将社区版 obcdc 接入自己的数据消费链路。 #### 功能适用性 该内容仅适用于社区版 obcdc。 ## 使用 obcdc 动态库开发自己的 OceanBase 数据消费工具 obcdc 是 C++ 编写，编译产出物为动态库，下游消费程序开发时需要依赖动态库和头文件（libobcdc.h、ob_errno.h）。 ### 引入 社区版 obcdc 依赖 oceanbase-ce-libs 包，可以通过 [软件中心](https://open.oceanbase.com/softwareCenter/community) 进行下载安装。 ```bash rpm -ivh oceanbase-ce-libs-****.rpm ``` 安装完成后执行 `ldd ./libobcdc.so` 命令，检查本地是否有缺少的动态库，如果有缺少，请保证所有依赖库都在本地，并为使用 libobcdc 的程序配置 `LD_LIBRARY_PATH` 保证 obcdc 可以正常连接，例如 `oceanbase-ce-libs` 中的 `libmariadb.so.3`。 ### 头文件 [libobcdc.h](https://github.com/oceanbase/oceanbase/blob/master/src/logservice/libobcdc/src/libobcdc.h) 中有详细的接口说明，您可以参考代码中的接口说明，本文 [附录](#附录) 中简单列出常用接口。 ### 构建 obcdc 实例 您可通过 `ObCDCFactory::construct_obcdc()` 方法构造 obcdc。 ### 使用 obcdc - 初始化接口：需要通过 `init/init_with_start_tstamp_usec` 接口通知 obcdc 配置信息和启动时间戳信息。 - 配置信息：支持传入配置文件路径、配置文件的 map。 - 指定启动时间（拉日志的起始时间）：时间单位可以是秒，也可以是毫秒。 - 启动接口：通过 launch 接口通知 obcdc 开始工作。 - 获取 LogRecord 的接口：通过 `next_record` 接口不断的从 obcdc 中拿到 OceanBase 的增量数据，这个接口可以指定获取超时时间，可以指定租户的数据。数据被封装为 LogRecord 格式，LogRecord 内存由 obcdc 分配。 - 归还 LogRecord 的接口：通过 `release_record` 接口将完成消费的 LogRecord 归还给 obcdc，obcdc 有后台的 GC 线程回收内存（异步回收）。 - 获取当前 obcdc 服务的所有租户 ID：通过 `get_tenant_ids` 获取当前 obcdc 服务中的所有租户列表。 ### 销毁 obcdc 实例 销毁 obcdc 要先停止 obcdc 实例，然后销毁 obcdc。 - 停止 obcdc： - 通过 `stop` 接口通知 obcdc 停止各模块的运行。 - 通过 `destroy` 接口析构 obcdc 的各模块，释放相关资源。 - 销毁 obcdc 实例：通过 `ObCDCFactory::deconstruct(IObCDCInstance *instance)` 销毁 obcdc 实例，此后，不能再访问第一步拿到的 obcdc 实例的指针。 ### 注意事项 - 从 obcdc 中取到的数据都是在 obcdc 进程内分配的内存，所以请务必保证 `next_record` 和 `release_record` 接口成对调用，避免内存无法释放。允许多次 `next_record` 后执行一起为对应的 LogRecord 调用 `release_record`。 - 调用 stop 和 launch 后不能再调用 launch/init 接口。 - 您需要对 obcdc 接口返回的所有错误码按需适当处理： - 返回错误码 `OB_SUCCESS` 表示取数据成功，返回的数据指针一定不为 NULL。 - 返回错误码 `OB_TIMEOUT` 表示目前在 obcdc 中拉不到数据，收到该错误码后可以重试获取数据。如果数据位点实时表示正常，如果数据位点不实时则可能 obcdc 内部存在问题，需要进一步排查。 - 返回 `OB_IN_STOP_STATE`，说明 obcdc 停止了，可能是调用方主动触发了 obcdc 的 stop/destroy 接口，也可能是 obcdc 内部无法处理的异常，各模块标记停止工作。收到这个错误码可以记录安全位点等必要信息后退出进程。 - 收到其它类型错误码都是非预期的，可以记录安全位点等必要信息后退出进程。 ## 示例 这里提供一个简单的 demo 展示如何使用 obcdc，详细信息，请参考 [obcdc_tailf](https://www.oceanbase.com/docs/common-oceanbase-database-cn-1000000005683201) 以及 [obcdc_tailf 源码](https://github.com/oceanbase/oceanbase/tree/master/src/logservice/libobcdc/tests)。 #### 注意 这个 demo 仅用于提供开发指引，不能执行编译或运行。 ```C++ /** * Copyright (c) 2021 OceanBase * OceanBase CE is licensed under Mulan PubL v2. * You can use this software according to the terms and conditions of the Mulan PubL v2. * You may obtain a copy of Mulan PubL v2 at: * http://license.coscl.org.cn/MulanPubL-2.0 * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, * EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, * MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE. * See the Mulan PubL v2 for more details. * * obcdc_demo */ #include #include "include/libobcdc/libobcdc.h" #include "include/libobcdc/ob_errno.h" using namespace std; using namespace oceanbase::libobcdc; using namespace oceanbase::common; typedef IBinlogRecord Record; #define LOG(msg) \ do { \ std::cout << msg << std::endl; \ } while (0) int create_obcdc_instance(ObCDCFactory &cdc_factory, IObCDCInstance *&obcdc_instance) { int ret = OB_SUCCESS; if (NULL == (obcdc_instance = cdc_factory.construct_obcdc())) { ret = OB_NOT_INIT; LOG("[ERROR] construct_obcdc failed"); } return ret; } void destroy_obcdc_instance(ObCDCFactory &cdc_factory, IObCDCInstance *obcdc_instance) { obcdc_instance->stop(); cdc_factory.deconstruct(obcdc_instance); } int init_obcdc_instance(IObCDCInstance &obcdc_instance) { int ret = OB_SUCCESS; const char *config_path = "conf/libobcdc.conf"; if (OB_SUCCESS != (ret = obcdc_instance.init(config_path, 0))) { LOG("obcdc_instance init failed"); } else if (OB_SUCCESS != (ret = obcdc_instance.launch())) { LOG("obcdc_instance launch failed"); } int fetch_next_cdc_record(IObCDCInstance &obcdc_instance, Record *record) { int ret = OB_SUCCESS; const int64_t timeout = 10000; // usec if (OB_SUCCESS != (ret = obcdc_instance.next_record(&record, timeout))) { if (OB_TIMEOUT != ret) { LOG("[WARN] next_record failed"); } } else if (NULL == record) { ret = OB_ERR_UNEXPECTED; LOG("invalid record"); } int release_cdc_record(IObCDCInstance &obcdc_instance, Record *record) { int ret = OB_SUCCESS; obcdc_instance.release_record(record); return ret; } int handle_cdc_record(Record *record) { int ret = OB_SUCCESS; return ret; } int main(int argc, char **argv) { int ret = OB_SUCCESS; ObCDCFactory cdc_factory; IObCDCInstance *obcdc_instance = NULL; if (OB_SUCCESS != create_obcdc_instance(cdc_factory, obcdc_instance)) { LOG("[ERROR] construct_obcdc_instance failed"); } else if (NULL == obcdc_instance) { ret = OB_ERR_UNEXPECTED; LOG("[ERROR] obcdc_instance should not be null!"); } else { if (OB_SUCCESS != init_obcdc_instance(*obcdc_instance)) { LOG("[ERROR] obcdc_instance init failed"); } else { while(OB_SUCCESS == ret) { Record *record = NULL; if (OB_SUCCESS != (ret = fetch_next_cdc_record(*obcdc_instance, record))) { if (OB_TIMEOUT == ret) { ret = OB_SUCCESS; } else { LOG("[ERROR] fetch_next_cdc_record failed"); } } else if (OB_SUCCESS != (ret = handle_cdc_record(record))) { LOG("[ERROR] handle_cdc_record failed"); } else if (OB_SUCCESS != (ret = release_cdc_record(*obcdc_instance, record))) { LOG("[ERROR] release_cdc_record failed"); } } } destroy_obcdc_instance(cdc_factory, obcdc_instance); } return 0; } ``` ## 附录 该章节主要展示部分 obcdc 头文件代码。 ### ObCDCFactory ObCDCFactory 是 obcdc 实例工厂，负责构建或销毁 obcdc 实例。 #### 注意 一个进程只能构造一个 obcdc 实例。 ```C++ // libobcdc.h: ObCDCFactory namespace oceanbase{ namespace liboblog{ class ObCDCFactory { public: ObCDCFactory(); ~ObCDCFactory(); public: IObCDCInstance *construct_obcdc(); void deconstruct(IObCDCInstance *log); }; } } ``` ### IObCDCInstance IObCDCInstance 是 obcdc 的对外接口，提供初始化、启动、停止、销毁、取数据、归还数据的能力，这里列出了一些常用的接口定义。 ```C++ // libobcdc.h: IObCDCInstance namespace oceanbase{ namespace liboblog{ // IObCDCInstance 是 liboblog 对外提供的接口 // 注意：这里忽略接口中需要传入的参数 class IObCDCInstance { public: virtual ~IObCDCInstance() {}; public: /* * init libobcdc * @param config_file config file name * @param start_timestamp start timestamp (by second) * @param err_cb error callback function pointer */ virtual int init(const char *config_file, const uint64_t start_timestamp_sec, ERROR_CALLBACK err_cb = NULL) = 0; /* * init libobcdc * @param configs config by map * @param start_timestamp start timestamp (by second) * @param err_cb error callback function pointer */ virtual int init(const std::map &configs, const uint64_t start_timestamp_sec, ERROR_CALLBACK err_cb = NULL) = 0; /* * init libobcdc * @param configs config by map * @param start_timestamp start timestamp by microsecond * @param err_cb error callback function pointer */ virtual int init_with_start_tstamp_usec(const std::map &configs, const uint64_t start_timestamp_usec, ERROR_CALLBACK err_cb = NULL) = 0; virtual void destroy() = 0; /* * fetch next binlog record from OB cluster * @param record binlog record, memory allocated by oblog, support release_record(corresponding times) after mutli next_record * @param OB_SUCCESS success * @param OB_TIMEOUT timeout * @param other errorcode fail */ virtual int next_record(ICDCRecord **record, const int64_t timeout_us) = 0; /* * fetch next binlog record from OB cluster * @param [out] record binlog record, memory allocated by oblog, support release_record(corresponding tiems) after mutli next_record * @param [out] major_version major version of ICDCRecord * @param [out] tenant_id tenant id of ICDCRecord * * @param OB_SUCCESS success * @param OB_TIMEOUT timeout * @param other error code fail */ virtual int next_record(ICDCRecord **record, int32_t &major_version, uint64_t &tenant_id, const int64_t timeout_us) = 0; /* * release recorcd for EACH ICDCRecord * @param record */ virtual void release_record(ICDCRecord *record) = 0; /* * Launch libobcdc * @retval OB_SUCCESS on success * @retval ! OB_SUCCESS on fail */ virtual int launch() = 0; /* * Stop libobcdc */ virtual void stop() = 0; /// get all serving tenant id list after oblog inited /// /// @param [out] tenant_ids tenant ids that oblog serving /// /// @retval OB_SUCCESS success /// @retval other value fail virtual int get_tenant_ids(std::vector &tenant_ids) = 0; }; } } ``` 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线