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

前提条件

• 已创建新版数据订阅通道，详情请参见创建RDS MySQL数据订阅通道（新版）、创建PolarDB MySQL数据订阅通道或创建Oracle数据订阅通道。
• 已创建一个或多个消费组，详情请参见新增消费组。

注意事项

• 使用本文提供的Demo消费数据时，如果采用auto commit（自动提交），可能会因为数据还没被消费完就执行了提交操作，从而丢失部分数据，建议采用手动提交的方式以避免该问题。
  说明 如果发生故障没有提交成功，重启客户端后会从上一个记录的位点进行数据消费，期间会有部分重复数据，您需要手动过滤。
• 数据以Avro序列化存储，详细格式请参见Record.avsc文档。
  说明 如果您使用的不是本文提供的客户端，您需要根据avro schema定义进行数据解析。

Kafka客户端Demo下载及运行流程说明

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

操作步骤

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

1. 下载Kafka客户端Demo代码，然后解压该文件。
2. 打开IntelliJ IDEA软件，然后单击Open。
   
3. 在弹出的对话框中，定位至Kafka客户端Demo代码下载的目录，参照下图依次展开文件夹，找到项目对象模型文件：pom.xml。
   
4. 在弹出对话框中，选择Open as Project。
5. 在IntelliJ IDEA软件界面，依次展开文件夹，找到并双击打开Kafka客户端Demo文件：NotifyDemo.java。
   
6. 设置NotifyDemo.java文件中的各参数对应的值。
   

   参数	说明	获取方式
   USER_NAME	消费组的账号。 警告 如果您使用的不是本文提供的客户端，请按照<消费组的的账号>-<消费组ID>的格式设置用户名（例如：dtstest-dtsae******bpv），否则无法正常连接。	在DTS控制台单击目标订阅实例ID，然后单击数据消费，您可以获取到消费组ID和消费组的账号信息。 说明 消费组账号的密码已在您新建消费组时指定。
   PASSWORD_NAME	该账号的密码。
   SID_NAME	消费组ID。
   GROUP_NAME	消费组名称，需保持和消费组ID相同（即本参数也填入消费组ID）。
   KAFKA_TOPIC	数据订阅通道的订阅Topic。	在DTS控制台单击目标订阅实例ID，在订阅配置页面，您可以获取到订阅Topic、网络地址及端口号信息。
   KAFKA_BROKER_URL_NAME	数据订阅通道的网络地址及端口号信息。 说明 如果您部署Kafka客户端所属的ECS实例与数据订阅通道属于经典网络或同一专有网络，建议通过内网地址进行数据订阅，网络延迟最小。
   INITIAL_CHECKPOINT_NAME	消费的数据时间点，格式为Unix时间戳。 说明 您需要自行保存时间点信息，当业务程序中断后，您可以通过订阅客户端传入已消费的数据时间点来继续消费数据，防止数据丢失。同时，您还可以在订阅客户端启动时，传入所需的消费位点，对订阅位点进行调整，实现按需消费数据。	首次订阅时，请自行将业务所需的时间点转化为Unix时间戳。
   USE_CONFIG_CHECKPOINT_NAME	默认取值为true，即强制使用指定的数据时间点来消费数据，避免丢失已接收到的但未处理的数据。	-

7. 在IntelliJ IDEA软件界面的顶部，选择Run > Run运行该客户端。
   说明 首次运行时，软件将自动加载相关依赖包并完成安装。

执行结果

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

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

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

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