创建数据订阅

前提条件

• 已将客户端IP添加至白名单中，具体操作请参见设置白名单。

• 已开通数据订阅功能，具体操作，请参见开通数据订阅。

• 确保源实例、目标实例和LTS服务网络互通，详细信息请参见网络互通。

• 已添加Lindorm宽表数据源，具体操作请参见Lindorm宽表数据源。

• 已添加Kafka数据源，具体操作请参见Kafka数据源。

通过Pull模式创建数据订阅通道

1. 进入LTS（原BDS）页面，在左侧导航栏中，选择数据订阅 > Pull模式。

   进入 Lindorm CDC 的 订阅通道列表 页面，页面右上角提供 创建数据订阅通道 按钮，列表展示已有通道的订阅通道 Id、Lindorm 表名、主题名及操作（详情、删除）。

2. 单击创建数据订阅通道，并配置以下参数。

   名称	描述
   源集群	填写Lindorm实例ID。
   Lindorm表名	选择需要创建数据订阅通道的Lindorm实例表，一条通道只能选择一张表格。
   主题名	用于消费数据的主题名称。
   数据过期时间（天）	表示数据可以保存的天数，默认为7天。
   主题分区数	表示Kafka客户端为该主题设置多个分区，多分区可以并发消费数据，默认为4个分区。

3. 单击提交。

4. （可选）找到目标订阅通道，单击操作列的详情，可以查看数据订阅通道详情、消费详情和存储详情。

5. （可选）您可以通过以下代码在Kafka客户端对订阅数据进行消费。

   import org.apache.hadoop.hbase.util.Bytes;
   import org.apache.kafka.clients.admin.AdminClientConfig;
   import org.apache.kafka.clients.consumer.ConsumerConfig;
   import org.apache.kafka.clients.consumer.ConsumerRecord;
   import org.apache.kafka.clients.consumer.ConsumerRecords;
   import org.apache.kafka.clients.consumer.KafkaConsumer;
   import org.apache.kafka.common.serialization.ByteArrayDeserializer;
   import java.time.Duration;
   import java.util.Arrays;
   import java.util.Properties;
   public class TestConsume {
     public static void main(String[] args) throws Exception {
       // 创建订阅通道时填写的topic名称
       String topic = "test-topic";
       // 链接endpoint的配置项
       Properties props = new Properties();
       // 指定链接endpoint地址
       props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "ld-xxx:9092");
       // 指定Key序列化器，不可更改
       props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // 指定Value序列化器，不可更改
       props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
       // 指定消费组名称，在消费时会自动创建
       props.put(ConsumerConfig.GROUP_ID_CONFIG, "group-id-0");
       // 创建消费者
       KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(props);
       // 订阅主题
       consumer.subscribe(Arrays.asList(topic));
       // 用消费者拉取数据
       ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(10000));
       for (ConsumerRecord<byte[], byte[]> record : records) {
         // 查看数据内容
         System.out.println("key: " + Bytes.toString(record.key()));
         System.out.println("value: " + Bytes.toString(record.value()));
       }
       // 提交当前消费位移
       consumer.commitSync();
       // 关闭消费者
       consumer.close();
     }
   }

   说明

   数据消费格式说明请参见数据消费格式。

通过Push模式创建数据订阅

推送流程

数据订阅功能将云原生多模数据库 Lindorm表格的增量数据消息主动推送至下游Kafka，推送流程如下。