tencent cloud

package com.qcloud.biz;
import com.qcloud.dts.context.NetworkEnv;
import com.qcloud.dts.context.SubscribeContext;
import com.qcloud.dts.message.ClusterMessage;
import com.qcloud.dts.message.DataMessage;
import com.qcloud.dts.subscribe.ClusterListener;
import com.qcloud.dts.subscribe.DefaultSubscribeClient;
import com.qcloud.dts.subscribe.SubscribeClient;
import java.util.List;
public class Main {
    public static void main(String[] args) throws Exception {
        //创建一个 context
        SubscribeContext context=new SubscribeContext();
        //用户 secretId、secretKey，建议使用子账号密钥，授权遵循最小权限指引，降低使用风险。子账号密钥获取可参考https://www.tencentcloud.com/document/product/598/32675
        context.setSecretId("AKID-522dabxxxxxxxxxxxxxxxxxx"); 
        context.setSecretKey("AKEY-0ff4cxxxxxxxxxxxxxxxxxxxx"); 
        // 设置 channel 所在的 region，2.8.0 以后的 SDK 必须设置 region 参数
        // region 值参照：https://www.tencentcloud.com/document/product/236/15833?from_cn_redirect=1#.E5.9C.B0.E5.9F.9F.E5.88.97.E8.A1.A8
        context.setRegion("ap-chongqing");
        // 订阅的 serviceIp 和 servicePort
        // 注意：2.8.0以前的 SDK 需要设置 IP 和 Port 两个参数，2.8.0以后的版本如果设置了 region 参数则可以省略
        // context.setServiceIp("10.xx.xx.24");
        // context.setServicePort(50120);

        // 如果运行 SDK 的 CVM 不能访问外网，设置网络环境为内网;默认为外网。
        context.setNetworkEnv(NetworkEnv.LAN);
        //创建客户端
        SubscribeClient client=new DefaultSubscribeClient(context);
        //创建订阅 listener
        ClusterListener listener= new ClusterListener() {
            @Override
            public void notify(List<ClusterMessage> messages) throws Exception {
                //消费订阅到的数据
                for(ClusterMessage m:messages){
                  for(DataMessage.Record.Field f:m.getRecord().getFieldList()){

                      if (f.getType().equals(DataMessage.Record.Field.Type.BLOB)){
                          System.out.println("["+f.getType()+"]["+f.getFieldname()+"] the original value:");
                          byte[] theRawBytesValue = f.getValueAsBytes();
                      }if(f.getType().equals(DataMessage.Record.Field.Type.INT8)){
                          // 如果该值为 null，f.getValueAsInteger() 返回 null
                          System.out.println(f.getValueAsInteger());
                      }if(f.getType().equals(DataMessage.Record.Field.Type.JSON)){
                          // 源实例为 mysql5.7 才可能返回 JSON 类型的数据
                          System.out.println(f.getValueAsString());
                      }if(f.getType().equals(DataMessage.Record.Field.Type.STRING)){
                          // 如果该值为 null，f.getValueAsString() 返回 null
                          System.out.println(f.getValueAsString());
                          // 字段原始的编码
                          System.out.println(f.getFieldEnc());
                      }
                      else{
                          // f.getValue() 方法即将废弃
                          String value = f.getValue() == null ? "Null": f.getValue();
                          String msg = "["+f.getType()+"]"+f.getFieldname()+"[encoding:"+f.getFieldEnc()+"]"+"[value:"+value+"]";
                          System.out.println(msg);
                      }

                  }
                  //消费完之后，确认消费
                  m.ackAsConsumed();
              }
            }
            @Override
            public void onException(Exception e){
                System.out.println("listen exception"+e);
            }};
        //添加监听者
        client.addClusterListener(listener);
        //设置请求的订阅通道
        client.askForGUID("dts-channel-r0M8kKsSyRZmSxQt");
        //启动客户端
        client.start();
    }
}