- 动态与公告
- 产品简介
- 购买指南
- 快速入门
- 准备工作
- 数据迁移
- 数据同步
- 数据订阅（Kafka 版）
- 前置校验不通过处理方法
- 割接说明
- 监控与告警
- 运维管理
- 实践教程
- 常见问题
- 错误处理
- API 文档
- History
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- ModifyMigrationJob
- CompleteMigrateJob
- DescribeMigrationDetail
- CreateMigrationService
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- StopMigrateJob
- ResumeMigrateJob
- RecoverMigrateJob
- ModifyMigrateName
- ModifyMigrateJobSpec
- IsolateMigrateJob
- DestroyMigrateJob
- DescribeMigrateDBInstances
- PauseMigrateJob
- ContinueMigrateJob
- Data Sync APIs
- DescribeSyncJobs
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- SkipSyncCheckItem
- DescribeCheckSyncJobResult
- StartSyncJob
- StopSyncJob
- ResumeSyncJob
- ResizeSyncJob
- RecoverSyncJob
- IsolateSyncJob
- DestroySyncJob
- PauseSyncJob
- ContinueSyncJob
- Data Consistency Check APIs
- Data Types
- Error Codes
- DTS API 2018-03-30
- SDK 文档
- 相关协议
- 动态与公告
- 产品简介
- 购买指南
- 快速入门
- 准备工作
- 数据迁移
- 数据同步
- 数据订阅（Kafka 版）
- 前置校验不通过处理方法
- 割接说明
- 监控与告警
- 运维管理
- 实践教程
- 常见问题
- 错误处理
- API 文档
- History
- Introduction
- API Category
- Making API Requests
- (NewDTS) Data Migration APIs
- DescribeMigrationJobs
- ModifyMigrationJob
- CompleteMigrateJob
- DescribeMigrationDetail
- CreateMigrationService
- CreateMigrateCheckJob
- DescribeMigrationCheckJob
- SkipCheckItem
- StartMigrateJob
- StopMigrateJob
- ResumeMigrateJob
- RecoverMigrateJob
- ModifyMigrateName
- ModifyMigrateJobSpec
- IsolateMigrateJob
- DestroyMigrateJob
- DescribeMigrateDBInstances
- PauseMigrateJob
- ContinueMigrateJob
- Data Sync APIs
- DescribeSyncJobs
- ModifySyncJobConfig
- CreateModifyCheckSyncJob
- DescribeModifyCheckSyncJobResult
- StartModifySyncJob
- CreateSyncJob
- ConfigureSyncJob
- CreateCheckSyncJob
- SkipSyncCheckItem
- DescribeCheckSyncJobResult
- StartSyncJob
- StopSyncJob
- ResumeSyncJob
- ResizeSyncJob
- RecoverSyncJob
- IsolateSyncJob
- DestroySyncJob
- PauseSyncJob
- ContinueSyncJob
- Data Consistency Check APIs
- Data Types
- Error Codes
- DTS API 2018-03-30
- SDK 文档
- 相关协议
功能概述
数据订阅是指获取数据库中关键业务的数据变化信息，将这些信息包装为消息对象推送到 Kafka 中，方便下游业务订阅、获取和消费。腾讯云 DTS 支持通过 Kafka/Flink Client 直接消费数据，方便用户搭建云数据库和异构系统之间的数据同步，如缓存更新，ETL（数据仓库技术）实时同步，业务异步解耦等。
实现原理
如下以 MySQL 为例进行介绍。数据订阅通过实时拉取源实例的 Binlog 增量日志，将增量数据解析成 Kafka message，然后存储到 Kafka Server。用户可以通过 Kafka Client 来消费数据，Kafka 作为一种开源的消息中间件，支持多数据通道消费和多种 SDK 语言，降低用户的使用成本。
![]()
典型应用场景
数据归档存储
使用 DTS 的数据订阅特性，您可将云数据库的增量更新数据实时地流式推送到归档数据库或数据仓库。
![]()
约束限制
订阅的消息内容目前默认保存时间为最近1天，超过保存时间的数据会被清除，请用户及时消费，避免数据在消费完之前就被清除。
数据消费的地域需要与订阅实例的地域相同。
对 MySQL、MariaDB、TDSQL MySQL 的数据订阅，当前不支持 geometry 相关的数据类型。
性能说明
订阅链路中，源库解析的数据先写入 DTS 内置 Kafka 中，然后通过客户端进行消费。写入和消费的性能如下：
场景 | 性能上限参考 |
数据写入内置 Kafka（MySQL/MariaDB/Percona/TDSQL-C MySQL/TDSQL MySQL 单分片） | 10MB/s |
数据写入内置 Kafka（TDSQL MySQL 多分片） | 10MB/s * 分片数 |
从内置 Kafka 消费数据 | 20MB/s（单消费组） |
| 50MB/s（多消费组） |
如上性能数据仅提供参考，因为在实际使用过程中，源数据库负载过大、网络延迟过高等因素都可能导致性能受限。
支持的订阅类型
DTS 支持订阅对象选择的粒度为库、表，具体支持如下三种订阅类型。
数据更新：指订阅 DML 操作。
结构更新：指订阅 DDL 操作。
全实例：指订阅所有库表的 DML 以及 DDL 操作。
消费支持的数据格式
用户对订阅数据进行消费时，支持 ProtoBuf、Avro 和 JSON 三种形式。ProtoBuf 及 Avro 是二进制格式，效率更高，JSON 为轻量级的文本格式，更加简单易用。
支持的高级功能
是
否
本页内容是否解决了您的问题？