- 动态与公告
- 产品简介
- 购买指南
- 快速入门
- 准备工作
- 数据迁移
- 数据同步
- 数据订阅（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 文档
- 相关协议
问题现象
数据订阅场景中，用户使用自己的消费程序进行数据消费，可能遇到如下异常：
1. 无法消费数据。
2. 消费到的数据丢失或者重复。
3. 消费端延迟越来越大。
排查思路
1. 无法消费数据
如果用户使用自己的消费端程序无法消费到数据，请先使用 DTS 提供的 Demo 进行消费测试。
如果使用 DTS 提供的 Demo 可以消费数据，则需要用户排查自己的消费程序。
如果使用 DTS 提供的 Demo 也无法消费数据，则需要进行如下排查：
检查消费端网络环境是否满足要求，消费端需要在腾讯云内网且与 DTS 订阅任务同地域。
检查启动 Demo 的参数填写是否有误，尤其是消费组密码。
检查 Demo 版本是否使用正确，不同的源库类型，不同的数据格式需要使用不同的 Demo 版本。
检查订阅任务是否有数据写入 Kafka（在控制台消费组管理中查看未消费的消息数是否大于零）。
2. 消费到的数据丢失或者重复
用户消费到的数据不符合预期，出现数据重复、丢失等问题，一般的原因是用户的消费程序处理异常，请用户先进行问题重现。具体复现问题的操作，参考如下二选一即可：
在控制台上调整 Kafka 位点，将位点回调到之前的位置再次消费。
新建消费组，在新的消费组上重复消费。（不同消费组上消费互不影响）
3. 消费端数据延迟越来越大
1. 消费端程序 commit 的逻辑问题。
如果消费程序只消费不做 commit，Kafka 的位点就不会更新。DTS Demo 中默认的 commit 逻辑为：当消费到 checkpoint 类型的消息时做一次 commit，订阅服务大概每10秒写入一个 checkpoint 类型的消息。如果用户修改了 commit 规则就有可能出现该问题，请用户先排查消费程序中的 commit 规则。
2. 消费效率太低。
消费效率受多个因素影响，网络情况、消费程序的处理效率、并发消费、多分区消费等。用户可以新建一个消费组用 DTS 提供的 Demo 消费，对比用户自己的消费程序的速度，排除消费程序的效率卡点。其他的就需要用户检查网络情况，提高数据处理速度或者增加消费者数量，多个分区的 topic 尽量使用多进程并发消费。
是
否
本页内容是否解决了您的问题？