創(chuàng)建數(shù)據(jù)通道

1. 登錄表格存儲控制臺。

2. 在概覽頁面，單擊實例名稱或在操作列單擊實例管理。

3. 在實例詳情頁簽的數(shù)據(jù)表列表區(qū)域，單擊數(shù)據(jù)表名稱后選擇實時消費通道頁簽或單擊后選擇實時消費通道。

4. 在實時消費通道頁簽，單擊創(chuàng)建通道。

5. 在創(chuàng)建通道對話框，輸入通道名，并選擇通道類型。

   通道服務提供了增量、全量、全量加增量三種類型的分布式數(shù)據(jù)實時消費通道。本文以全量加增量類型為例介紹。

   創(chuàng)建成功后，在操作列單擊展示通道分區(qū)列表，可以查看通道中的數(shù)據(jù)內(nèi)容、消費延遲監(jiān)控以及通道分區(qū)下的消費數(shù)據(jù)行數(shù)統(tǒng)計。

預覽通道中的數(shù)據(jù)格式

創(chuàng)建通道后，通過模擬數(shù)據(jù)消費可以預覽通道中的數(shù)據(jù)格式。

1. 寫入或刪除數(shù)據(jù)，詳情請參見控制臺讀寫數(shù)據(jù)。

2. 預覽通道中的數(shù)據(jù)格式。

   1. 在概覽頁頁面，單擊實例名稱或在操作列單擊實例管理。

   2. 在實例詳情頁簽的數(shù)據(jù)表列表區(qū)域，單擊數(shù)據(jù)表名稱后選擇實時消費通道頁簽或單擊后選擇實時消費通道。

   3. 在實時消費通道頁簽，單擊通道操作列的展示通道分區(qū)列表。

   4. 在通道分區(qū)的右側(cè)單擊模擬消費。

   5. 在模擬消費對話框，單擊開始消費。

      消費的數(shù)據(jù)信息顯示在對話框中，如下圖所示。

開啟通道的數(shù)據(jù)消費

1. 在通道列表中復制通道ID。

2. 使用任一種語言的通道SDK，開啟通道的數(shù)據(jù)消費。

   此處以Java SDK為例開啟通道的數(shù)據(jù)消費。

   import com.alicloud.openservices.tablestore.TunnelClient;
   import com.alicloud.openservices.tablestore.tunnel.worker.IChannelProcessor;
   import com.alicloud.openservices.tablestore.tunnel.worker.ProcessRecordsInput;
   import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorker;
   import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorkerConfig;
   
   public class TunnelQuickStart {
   
       /**
        * 用戶自定義數(shù)據(jù)消費Callback，即實現(xiàn)IChannelProcessor接口（process和shutdown）。
        */
       private static class SimpleProcessor implements IChannelProcessor {
           @Override
           public void process(ProcessRecordsInput input) {
               System.out.println("Default record processor, would print records count");
               System.out.println(
                       String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
               try {
                   //模擬消費處理。
                   Thread.sleep(1000);
               } catch (InterruptedException e) {
                   e.printStackTrace();
               }
           }
   
           @Override
           public void shutdown() {
               System.out.println("Mock shutdown");
           }
       }
   
       public static void main(String[] args) {
           // 1. 初始化Tunnel Client。
           TunnelClient tunnelClient = new TunnelClient("<ENDPOINT>", "<ACCESS_ID>", "<ACCESS_KEY>", "<INSTANCE_NAME>");
   
           // 2. 用戶自定義數(shù)據(jù)消費Callback, 開始自動化的數(shù)據(jù)消費。
           //強烈建議共用一個TunnelWorkerConfig，TunnelWorkerConfig中包括更多的高級參數(shù)。
           TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
           //配置TunnelWorker，并啟動自動化的數(shù)據(jù)處理任務。
           //TUNNEL_ID，通道id。
           TunnelWorker worker = new TunnelWorker("<TUNNEL_ID>", tunnelClient, config);
           try {
               worker.connectAndWorking();
           } catch (Exception e) {
               e.printStackTrace();
               worker.shutdown();
               tunnelClient.shutdown();
           }
       }
   }

查看數(shù)據(jù)消費日志

數(shù)據(jù)消費后，可以查看增量數(shù)據(jù)消費日志，例如消費統(tǒng)計、增量通道分區(qū)最新同步時間等。在控制臺或者使用describeTunnel接口也可以查看消費延遲、通道分區(qū)下的消費數(shù)據(jù)行數(shù)更新。