content: package com.aliyun.dts.demo;
  import java.lang.reflect.Field;
  import java.util.List;
  import java.util.Iterator;
  import com.fasterxml.jackson.databind.JsonNode;
  import com.alicloud.openservices.tablestore.ClientException;
  import com.alicloud.openservices.tablestore.SyncClient;
  import com.alicloud.openservices.tablestore.TableStoreException;
  import com.alicloud.openservices.tablestore.model.*;
  import com.alicloud.openservices.tablestore.model.condition.ColumnCondition;
  import com.alicloud.openservices.tablestore.model.condition.SingleColumnValueCondition;
  import com.alicloud.openservices.tablestore.model.filter.SingleColumnValueFilter;
  
  import com.aliyun.drc.client.message.DataMessage;
  import com.aliyun.drc.clusterclient.ClusterClient;
  import com.aliyun.drc.clusterclient.message.ClusterMessage;
  import com.aliyun.drc.clusterclient.DefaultClusterClient;
  import com.aliyun.drc.clusterclient.RegionContext;
  import com.aliyun.drc.clusterclient.ClusterListener;
  
  import com.aliyun.dts.demo.utils.Utils;
  
  public class Main {
  
      private static final String TABLE_NAME = ""; //Tablestore tablename
  
      public static void main(String[] args) {
          JsonNode config = Utils.getClientAndConfig();
          final String accessId = config.get("accessId").asText();
          final String accessKey = config.get("accessKey").asText();
          final String otsEndPoint = config.get("endpoint").asText();
          final String otsInstanceName = config.get("instanceName").asText();
          final String dtsGuid = config.get("dtsguid").asText();//DTS inanse ID
  
          final SyncClient otsclient = new SyncClient(otsEndPoint, accessId, accessKey, otsInstanceName);
  
          // create a context
          RegionContext context = new RegionContext();
          // connect to dts by  internet
          context.setUsePublicIp(true);
          // aliyun accessKey secret
          context.setAccessKey(accessId);
          context.setSecret(accessKey);
          // create a cosummer
          final ClusterClient client = new DefaultClusterClient(context);
  
          // create listener
          ClusterListener listener = new ClusterListener() {
              @Override
              public void notify(List<ClusterMessage> messages) throws Exception {
                  for (ClusterMessage message : messages) {
                      // debug
                      String Opt = message.getRecord().getOpt().toString();
                      List<DataMessage.Record.Field> FieldList = message.getRecord().getFieldList();
                      System.out.println(FieldList);
                      Long[] field = new Long[10];
                      for (int i = 0; i < FieldList.size(); i++) {
                          field[i] =  Long.parseLong(FieldList.get(i).getValue().toString());
                      }
                      //System.out.println(field);
                      System.out.println(Opt);
                      if (Opt == "INSERT") {
                          System.out.println("a");
                          int num = FieldList.size();
                          try {
                              putRow(otsclient, num, field);
                          } catch (TableStoreException e) {
                              System.err.println("操作失败，详情：" + e.getMessage());
                              System.err.println("Request ID:" + e.getRequestId());
                          } catch (ClientException e) {
                              System.err.println("请求失败，详情：" + e.getMessage());
                          }
  
                      } else if (Opt == "DELETE") {
                          System.out.println("b");
                          try {
                              deleteRow(otsclient,field);
                          } catch (TableStoreException e) {
                              System.err.println("操作失败，详情：" + e.getMessage());
                              System.err.println("Request ID:" + e.getRequestId());
                          } catch (ClientException e) {
                              System.err.println("请求失败，详情：" + e.getMessage());
                          }
                      } else if (Opt == "UPDATE") {
                          System.out.println("c");
                          Long[] field0 = new Long[5];
                          Long[] field1 = new Long[5];
                          int num = FieldList.size() / 2;
                          for(int i=0; i<num; i++) {
                              field0[i] = field[i * 2];
                              field1[i] = field[i * 2 + 1];
                          }
                          try {
                              deleteRow(otsclient,field0);
                              putRow(otsclient,num, field1);
                          } catch (TableStoreException e) {
                              System.err.println("操作失败，详情：" + e.getMessage());
                              System.err.println("Request ID:" + e.getRequestId());
                          } catch (ClientException e) {
                              System.err.println("请求失败，详情：" + e.getMessage());
                          }
                      }
                      //you must call ackAsConsumed when you consume the data
                      message.ackAsConsumed();
                  }
              }
  
              @Override
              public void noException(Exception e) {
                  e.printStackTrace();
              }
          };
          // add listener
          client.addConcurrentListener(listener);
          // set inanse ID
          try{
              client.askForGUID(dtsGuid);
              client.start();
          }catch ( Exception e) {
              System.err.println("请求失败，详情：" + e.getMessage());
          }
  
          //otsclient.shutdown();
  
      }
  
      private static void putRow(SyncClient client, int num, Long field[]) {
          // 构造主键
          PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
          primaryKeyBuilder.addPrimaryKeyColumn("pk1",PrimaryKeyValue.fromLong(field[0]));
          primaryKeyBuilder.addPrimaryKeyColumn("pk2",PrimaryKeyValue.fromLong(field[1]));
          PrimaryKey primaryKey = primaryKeyBuilder.build();
          RowPutChange rowPutChange = new RowPutChange(TABLE_NAME, primaryKey);
          //加入属性列
          for(int i=2; i<num ;i++)
              rowPutChange.addColumn(new Column("v"+i, ColumnValue.fromLong(field[i])));
  
          client.putRow(new PutRowRequest(rowPutChange));
      }
  
      private static void deleteRow(SyncClient client, Long pk[]) {
          // 构造主键
          PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
          primaryKeyBuilder.addPrimaryKeyColumn("pk1",PrimaryKeyValue.fromLong(pk[0]));
          primaryKeyBuilder.addPrimaryKeyColumn("pk2",PrimaryKeyValue.fromLong(pk[1]));
          PrimaryKey primaryKey = primaryKeyBuilder.build();
  
          RowDeleteChange rowDeleteChange = new RowDeleteChange(TABLE_NAME, primaryKey);
  
          client.deleteRow(new DeleteRowRequest(rowDeleteChange));
      }
  
  }
  