--- title: "obcdc 开发说明 | OceanBase 文档中心" description: obcdc 开发说明 本文介绍如何将 obcdc 接入自己的数据消费链路。 使用 obcdc 动态库开发自己的 OceanBase 数据消费工具 obcdc 是 C++ 编写，编译产出物为动态库，下游消费程序开发时需要依赖动态库和头文件（libobcdc.h、ob_errno.h）。 引入 执行 ldd ./libo… image: https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*OSPzQ6GUQF4AAAAAQHAAAAgAeiGDAQ/original --- 切换语言 - 中文站 - 简体中文 - International - English - 日本站 - 日本語 文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*P8CuR4UJ_FkAAAAAAAAAAAAADiGDAQ/original) OceanBase 数据库分布式版 - V 3.1.4 社区版 # obcdc 开发说明 更新时间：2025-01-20 15:35:17 [编辑](https://github.com/oceanbase/oceanbase-doc/edit/V3.1.4/zh-CN/1800.supporting-tools/400.cdc/200.obcdc/100.deploy-and-use-obcdc/200.obcdc-development-instructions.md) 本文介绍如何将 obcdc 接入自己的数据消费链路。 ## 使用 obcdc 动态库开发自己的 OceanBase 数据消费工具 obcdc 是 C++ 编写，编译产出物为动态库，下游消费程序开发时需要依赖动态库和头文件（libobcdc.h、ob_errno.h）。 ### 引入 执行 `ldd ./libobcdc.so` 看本地是否有缺少的动态库，如果有缺少，请保证所有依赖库（比如 libdrcmessage.so（商业版））都在本地，并配置 `LD_LIBRARY_PATH` 保证 obcdc 能链接到。 ### 头文件 [libobcdc.h](https://github.com/oceanbase/oceanbase/blob/master/tools/obcdc/src/libobcdc.h) 中有详细的接口说明，您可以参考代码中的接口说明，本文 **附录** 中简单列出常用接口。 ### 构建 obcdc 实例 您可通过 `ObLogFactory::construct_oblog()` 方法构造 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 实例：通过 `ObLogFactory::deconstruct(IObLog *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/community-observer-cn-10000000000450603) 以及 [obcdc_tailf 源码](https://github.com/oceanbase/oceanbase/tree/master/tools/obcdc/tests)。 #### 注意 这个 demo 仅用于提供开发指引，不能执行编译或运行。 ```C++ #include "libobcdc.h" #include "ob_errno.h" using namespace oceanbase::common; using namespace oceanbase::liboblog; namespace oceanbase { namespace liboblog { class ObLogDemo { public: static ObLogDemo& get_instance(); static const int64_t NEXT_RECORD_TIMEOUT = 1000000; public: ObLogDemo(); ~ObLogDemo(); public: int init(const char* config_file_path, const uint64_t start_tstamp); void destroy(); int start(); void stop(); void run(); private: int handle_log_record(ILogRecord *log_record, volatile bool stop_flag); private: ObLogFactory oblog_factory_; IObLog* oblog_; bool inited_; volatile bool stop_flag_; }; } } ``` ### oblog_demo.cpp ```C++ #include "oblog_demo.h" #include "ob_log_module.h" // ONLY FOR OB LOG MODULE, DEVELOPER CAN REPLACE IT AND LOG_XXX BELOW WITY YOUR OWN LOG MODULE #define USING_LOG_PREFIX OBLOG_TAILF // ONLY FOR LOG ObLogDemo& ObLogDemo::get_instance() { static ObLogDemo oblog_instance; return oblog_instance; } ObLogDemo::ObLogDemo() : oblog_factory_(), oblog_(NULL), inited_(false), stop_flag_(true) {} ObLogDemo::~ObLogDemo() { destroy(); } int ObLogDemo::init(const char* config_file_path, const uint64_t start_tstamp) { int ret = OB_SUCCESS; if (NULL == (oblog_ = oblog_factory_.construct_oblog())) { LOG_ERROR("failed to construct liboblog", KR(ret)); } else if (OB_SUCCESS != oblog_->init(config_file_path, start_tstamp)) { LOG_ERROR("failed to init liboblog instance", KR(ret), K(config_file_path, K(start_tstamp))); } else { inited_ = true; } return ret; } void ObLogDemo::destroy() { stop(); inited_ = false; if (NULL != oblog_) { oblog_->destroy(); // will call ObLog::stop for another time oblog_factory_.deconstruct(oblog_); oblog_ = NULL; } } int ObLogDemo::start() { int ret = OB_SUCCESS; if (!inited_) { ret = OB_NOT_INIT; LOG_ERROR("ObLogDemo not init, please call init function first!", KR(ret)); } else if (!stop_flag_){ ret = OB_ERR_UNEXPECTED; LOG_ERROR("ObLogDemo already start!", KR(ret), K_(inited), K_(stop_flag)); } else { if (NULL == oblog_) { ret = OB_ERR_UNEXPECTED; LOG_ERROR("oblog instance should not be null while start ObLogDemo", KR(ret), K_(inited), K_(stop_flag)); } else if (OB_SUCCESS != oblog_->launch()) { LOG_ERROR("liboblog instance launch failed", KR(ret)); } else { LOG_INFO("liboblog instance launch success"); stop_flag_ = false; } } void ObLogDemo::stop() { stop_flag_ = true; if (NULL != oblog_) { oblog_->stop(); } } void ObLogDemo::run() { int ret = OB_SUCCESS; if (inited_ && NULL != oblog_) { while (OB_SUCCESS == ret && ! stop_flag_) { ILogRecord* record = NULL; if (OB_SUCCESS == oblog_->next_record(&record, NEXT_RECORD_TIMEOUT)) { if (OB_SUCCESS != handle_log_record(record, stop_flag_)) { LOG_ERROR("failed to handle LogRecord", KR(ret), K(record)); } // must call release_record oblog_->release_record(record); record = NULL; } else if (OB_TIMEOUT == ret) { LOG_WARN("fetch record timeout, may don't have any record", KR(ret)); // fetch record timeout may cause by no data produced by OB // process this situation with your own logic // e.g. retry or exit progress, etc. } else if (OB_IN_STOP_STATE == ret) { stop_flag_ = true; ret = OB_SUCCESS; } else { LOG_ERROR("next_record fail", KR(ret)); stop_flag_ = true; } } } else { LOG_ERROR("should not run ObLogDemo while not inid or oblog instance is NULL, will exit"); } } int ObLogDemo::handle_log_record(ILogRecord *log_record, volatile bool stop_flag) { int ret = OB_SUCCESS; // do your process for log_record // u can get stop signal by volatile variable stop_flag if (!stop_flag_) { LOG_DEBUG("handle log record", K(log_record)); } return ret; } } } ``` ## 附录 该章节主要给出部分 obcdc 头文件代码。 ### ObLogFactory ObLogFactory 是 obcdc 实例工厂，负责构建或销毁 obcdc 实例。 > **注意** > > > > 一个进程只能构造一个 obcdc 实例。 ```C++ // libobcdc.h: ObLogFactory namespace oceanbase{ namespace liboblog{ class ObLogFactory { public: ObLogFactory(); ~ObLogFactory(); public: // 构建 liboblog 实例 IObLog *construct_oblog(); // 销毁 liboblog 实例 void deconstruct(IObLog *log); }; } } ``` ### IObLog IObLog 是 obcdc 的对外接口，提供初始化、启动、停止、销毁、取数据、归还数据的能力，这里列出了一些常用的接口定义。 ```C++ // libobcdc.h: IObLog namespace oceanbase{ namespace liboblog{ // IObLog 是 liboblog 对外提供的接口 // 注意：这里忽略接口中需要传入的参数 class IObLog { public: virtual ~IObLog() {}; public: /* * init liboblog * @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, ERROR_CALLBACK err_cb = NULL) = 0; /* * init liboblog * @param configs config by map * @param start_timestamp start timestamp (by secon) * @param err_cb error callback function pointer */ virtual int init(const std::map& configs, const uint64_t start_timestamp, ERROR_CALLBACK err_cb = NULL) = 0; /* * init liboblog * @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(ILogRecord **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 ILogRecord * @param [out] tenant_id tenant id of ILogRecord * * @param OB_SUCCESS success * @param OB_TIMEOUT timeout * @param other error code fail */ virtual int next_record(ILogRecord **record, int32_t &major_version, uint64_t &tenant_id, const int64_t timeout_us) = 0; /* * release recorcd for EACH ILogRecord * @param record */ virtual void release_record(ILogRecord *record) = 0; /* * Launch liboblog * @retval OB_SUCCESS on success * @retval ! OB_SUCCESS on fail */ virtual int launch() = 0; /* * Stop liboblog */ 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) 咨询热线