765      微信

SDK快速入门__数据订阅_用户指南_数据传输-阿里云

这一节，您将学到如何用DTS Java SDK完成一些基本的操作。

初始化RegionContext

RegionContext 主要用于保存设置安全认证信息及访问网络模式设置。下面代码显示如何初始化RegionContext，设置安全认证凭证及网络访问模式。

import java.util.List;
import com.aliyun.drc.clusterclient.RegionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MainClass
{
   public static void main(String[] args) throws Exception {
      // 创建一个RegionContext
      RegionContext context = new RegionContext();
      // 配置阿里云账号的AccessKey及AccessKeySecret
      context.setAccessKey("<AccessKey>");
      context.setSecret("<AccessKeySecret>");
       // 运行SDK的服务器是否使用公网IP连接DTS订阅通道
      context.setUsePublicIp(true);
     // 下面为其他调用代码 ……
        …………
    }
}

如果要使用SDK，必须先初始化RegionContext，配置连接订阅通道的安全认证等信息。上面的接口setAccessKey设置的是阿里云账号的AccessKey
setSecret 设置的是阿里云账号的AccessKeySecret
AccessKey及AccessKeySecret是由阿里云的系统直接分配给用户的，称为ID对，用户标识用户，可到阿里云用户中心创建获取。
setUsePublicIp 是告诉DTS，您本地SDK运行服务器是否用公网IP连接订阅通道。如果设置为true，那么订阅数据流走公网，否则走内网。

初始化ClusterClient

SDK连接订阅通道，接受增量数据等操作都是通过类ClusterClient来完成的，下面代码创建了一个ClusterClient

import java.util.List;
import com.aliyun.drc.clusterclient.ClusterClient;
import com.aliyun.drc.clusterclient.DefaultClusterClient;
import com.aliyun.drc.clusterclient.RegionContext;
public class MainClass
{
      public static void main(String[] args) throws Exception {
        // 创建一个RegionContext
          RegionContext context = new RegionContext();
          context.setAccessKey("<AccessKey>");
          context.setSecret("<AccessKeySecret>");
          context.setUsePublicIp(true);
          // 创建订阅消费者
          final ClusterClient client = new DefaultClusterClient(context);
          // 下面是一些其他调用代码
          ……………
    }
}

import com.aliyun.drc.clusterclient.ClusterClient;
import com.aliyun.drc.clusterclient.ClusterListener;
import com.aliyun.drc.clusterclient.DefaultClusterClient;
import com.aliyun.drc.clusterclient.RegionContext;
import com.aliyun.drc.clusterclient.message.ClusterMessage;
public class MainClass
{
    public static void main(String[] args) throws Exception {
        // 初始化一个RegionContext对象
        ………
        //初始化一个ClusterClient对象
        ………
        ClusterListener listener = new ClusterListener(){
             @Override
             public void notify(List<ClusterMessage> messages) throws Exception {
                  for (ClusterMessage message : messages) {  
                    //打印订阅到的增量数据
                      System.out.println(message.getRecord() + ":" + message.getRecord().getTablename() + ":"
                      + message.getRecord().getOpt());  
                      //消费完数据后向DTS汇报ACK，必须调用
                      message.ackAsConsumed();
              }
      }
     }
}

import java.util.List;
import com.aliyun.drc.clusterclient.ClusterClient;
import com.aliyun.drc.clusterclient.ClusterListener;
import com.aliyun.drc.clusterclient.DefaultClusterClient;
import com.aliyun.drc.clusterclient.RegionContext;
import com.aliyun.drc.clusterclient.message.ClusterMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class MainClass
{
   public static void main(String[] args) throws Exception {
    //初始化RegionContext
    …………
    //初始化ClusterClient
    …………
    //初始化ClusterListener
    …………
    // 添加监听者
      client.addConcurrentListener(listener);
      // 设置请求的订阅通道ID
      client.askForGUID("dts_rdsrjiei2u2afnb_DSF");
      // 启动后台线程， 注意这里不会阻塞， 主线程不能退出
      client.start();
}