使用Kafka客户端消费订阅数据

注意事项

• 使用本文提供的Demo消费数据时，如果采用auto commit（自动提交），可能会因为数据还没被消费完就执行了提交操作，从而丢失部分数据，建议采用手动提交的方式以避免该问题。

  说明

  如果发生故障没有提交成功，重启客户端后会从上一个记录的位点进行数据消费，期间会有部分重复数据，您需要手动过滤。

• 数据以Avro序列化存储，详细格式请参见Record.avsc文档。

  警告

  如果您使用的不是本文提供的Kafka客户端，在进行反序列化解析（DTS Avro的反序列化示例）时，可能出现解析的数据有误，您需要自行验证数据的正确性。

• 关于offsetForTimes接口，DTS的搜索单位为秒，原生Kafka的搜索单位为毫秒。

• 由于数据订阅服务端会因容灾等原因导致网络闪断，若您未使用本文提供的Kafka客户端，您使用的Kafka客户端需具备网络重试能力。

• 若您使用原生的Kafka客户端消费订阅数据，则可能会在DTS发生增量数据采集模块切换行为，从而使subscribe模式下订阅客户端保存在服务端的消费位点被清除，您需要手动调整订阅的消费位点以实现按需消费数据。

Kafka客户端运行流程说明

请下载Kafka客户端Demo代码。更多关于代码使用的详细介绍，请参见Demo中的Readme文档。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.0.0</version>
</dependency>

操作步骤

本文以IntelliJ IDEA软件（Community Edition 2018.1.4 Windows版本）为例，介绍如何运行该客户端消费订阅通道中的数据。

1. 创建新版数据订阅通道，详情请参见创建RDS MySQL数据订阅通道、创建PolarDB MySQL版数据订阅通道或创建Oracle数据订阅通道。

2. 创建一个或多个消费组，详情请参见新增消费组。

3. 下载Kafka客户端Demo代码，然后解压该文件。

   说明

   单击，然后选择Download ZIP下载文件。

4. 打开IntelliJ IDEA软件，然后单击Open。

5. 在弹出的对话框中，定位至Kafka客户端Demo代码下载的目录，依次展开文件夹，找到项目对象模型文件：pom.xml。

   展开路径为 kafkademo > subscribe_example-master > javaimpl，选中 pom.xml 后单击 OK。

6. 在弹出对话框中，选择Open as Project。

7. 在IntelliJ IDEA软件界面，依次展开文件夹，找到并双击打开Kafka客户端Demo文件：NotifyDemoDB.java。

8. 设置NotifyDemoDB.java文件中的各参数对应的值。

   public static Properties getConfigs() {
       Properties properties = new Properties();
       // user password and sid for auth
       properties.setProperty(USER_NAME, "dtstest");
       properties.setProperty(PASSWORD_NAME, "xxx");
       properties.setProperty(SID_NAME, "dtsxxx");
       // kafka consumer group general same with sid
       properties.setProperty(GROUP_NAME, "dtsxxx");
       // topic to consume, partition is 0
       properties.setProperty(KAFKA_TOPIC, "cn_hangzhou_xxx");
       // kafka broker url
       properties.setProperty(KAFKA_BROKER_URL_NAME, "dts-cn-xxx.com:18001");
       // initial checkpoint for first seek(a timestamp to set, eg 1566180200 if you want (Mon Aug 19 10:03:21 CST 2019))
       properties.setProperty(INITIAL_CHECKPOINT_NAME, "1583307907");
       // if force use config checkpoint when start. for checkpoint reset
       properties.setProperty(USE_CONFIG_CHECKPOINT_NAME, "true");
       // use consumer assign or subscribe interface
       // when use subscribe mode, group config is required. kafka consumer group is enabled
       properties.setProperty(SUBSCRIBE_MODE_NAME, "assign");
       return properties;
   }

   参数	说明	获取方式
   USER_NAME	消费组的账号。 警告 如您未使用本文提供的客户端，请按照<消费组的账号>-<消费组ID>的格式设置用户名（例如：dtstest-dtsae******bpv），否则无法正常连接。	在DTS控制台单击目标订阅实例ID，然后单击数据消费，您可以获取到消费组ID和消费组的账号信息。 说明 消费组账号的密码已在您新建消费组时指定。
   PASSWORD_NAME	该账号的密码。
   SID_NAME	消费组ID。
   GROUP_NAME	消费组名称，需保持和消费组ID相同（即本参数也填入消费组ID）。
   KAFKA_TOPIC	数据订阅通道的订阅Topic。	在DTS控制台单击目标订阅实例ID，在任务管理页面，您可以获取到订阅Topic、网络地址信息。在DTS数据订阅任务详情页的基本信息区获取订阅Topic值，在网络区获取VPC网络地址（格式示例：xxx.aliyuncs.com:18003）。
   KAFKA_BROKER_URL_NAME	数据订阅通道的网络地址信息。 说明 如果您部署Kafka客户端所属的ECS实例与数据订阅通道属于经典网络或同一专有网络，建议通过内网地址进行数据订阅，网络延迟最小。 鉴于网络稳定性因素，不建议使用公网地址。
   INITIAL_CHECKPOINT_NAME	消费的数据时间点，格式为Unix时间戳，例如1592269238。 说明 您需要自行保存时间点信息，以便： 当业务程序中断后，传入已消费的数据时间点继续消费数据，防止数据丢失。 在订阅客户端启动时，传入所需的消费位点，调整订阅位点，实现按需消费数据。 若SUBSCRIBE_MODE_NAME为subscribe时，传入的INITIAL_CHECKPOINT_NAME仅在订阅客户端首次启动时生效。	消费的数据时间点必须在订阅实例的数据范围之内，并需转化为Unix时间戳。在 DTS 订阅任务列表中，可查看对应订阅任务的数据范围字段，该字段显示当前可消费数据的起止时间，据此可确定 INITIAL_CHECKPOINT_NAME 的有效取值范围。 说明 Unix时间戳转换工具可用搜索引擎获取。
   USE_CONFIG_CHECKPOINT_NAME	默认取值为true，即强制使用指定的数据时间点来消费数据，避免丢失已接收到的但未处理的数据。	无
   SUBSCRIBE_MODE_NAME	一个消费组下支持同时启动两个及以上Kafka客户端，如需实现该功能，请将所有客户端该参数的值设置为subscribe。 默认值为assign，即不使用该功能，建议只部署一个客户端。	无

9. 在IntelliJ IDEA软件界面的顶部，选择Run > Run运行该客户端。

   说明

   首次运行时，软件需要一定时间自动加载相关依赖包并完成安装。

执行结果

运行结果该客户端可正常订阅到源库的数据变更信息。

[2020-03-09 10:41:52,408] INFO [Consumer clientId=consumer-1, groupId=dts_xxx] Discovered coordinator xxx (id: xxx rack: null) (org.apache.kafka.clients.consumer.internals.AbstractCoordinator)
[2020-03-09 10:41:57,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721711, offset: 1732521, info: 1583721711] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:41:57,571] INFO EtlRecordProcessor: haven't receive records from generator for  5s (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:02,203] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721721, offset: 1732539, info: 1583721721] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:07,204] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721726, offset: 1732544, info: 1583721726] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:12,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721731, offset: 1732548, info: 1583721731] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:17,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721736, offset: 1732554, info: 1583721736] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:22,205] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721741, offset: 1732559, info: 1583721741] (recordprocessor.EtlRecordProcessor)
[2020-03-09 10:42:27,206] INFO commit record with checkpoint Checkpoint[ topicPartition: cn_hangzhou_rm_xxx_dtstest-0timestamp: 1583721746, offset: 1732569, info: 1583721746] (recordprocessor.EtlRecordProcessor)

您也可以去除NotifyDemoDB.java文件中的打印日志详情的注释（即删除第25行//log.info(ret);中的//），然后再次运行该客户端即可查看详细的数据变更信息。

常见问题

• Q：为什么需要自行记录客户端的消费位点？

  A：由于DTS记录的消费位点是接收到Kafka消费客户端执行commit操作的时间点，可能与当前实际消费到的时间点存在一定的时间差。当业务程序或Kafka消费客户端异常中断后，您可以传入自行记录的消费位点以继续消费，避免消费到重复的数据或缺失部分数据。

MySQL字段类型与dataTypeNumber数值的对应关系

Oracle字段类型与dataTypeNumber数值的对应关系

PostgreSQL字段类型与dataTypeNumber数值的对应关系