創建數據通道

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

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

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

4. 在實時消費通道頁簽，單擊創建通道。

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

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

   創建成功后，在操作列單擊展示通道分區列表，可以查看通道中的數據內容、消費延遲監控以及通道分區下的消費數據行數統計。

預覽通道中的數據格式

創建通道后，通過模擬數據消費可以預覽通道中的數據格式。

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

2. 預覽通道中的數據格式。

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

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

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

   4. 在通道分區的右側單擊模擬消費。

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

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

開啟通道的數據消費

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

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

   此處以Java SDK為例開啟通道的數據消費。

   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 {
   
       /**
        * 用戶自定義數據消費Callback，即實現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. 用戶自定義數據消費Callback, 開始自動化的數據消費。
           //強烈建議共用一個TunnelWorkerConfig，TunnelWorkerConfig中包括更多的高級參數。
           TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
           //配置TunnelWorker，并啟動自動化的數據處理任務。
           //TUNNEL_ID，通道id。
           TunnelWorker worker = new TunnelWorker("<TUNNEL_ID>", tunnelClient, config);
           try {
               worker.connectAndWorking();
           } catch (Exception e) {
               e.printStackTrace();
               worker.shutdown();
               tunnelClient.shutdown();
           }
       }
   }

查看數據消費日志

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