Table Store Java SDK 建立通道 - Tablestore

前提條件

安裝Tablestore Java SDK並初始化 TunnelClient。
已建立資料表。

功能說明

調用 createTunnel 為資料表建立通道。同一資料表可以建立多個通道。通道類型決定消費的資料範圍：BaseData 僅消費全量資料，Stream 僅消費增量資料，BaseAndStream 先消費全量資料，再消費增量資料。

說明

建立 Stream 或 BaseAndStream 類型的通道時，如果資料表未開啟 Stream，系統會自動開啟 Stream，並將增量日誌到期時間設定為 7 天。

CreateTunnelResponse createTunnel(CreateTunnelRequest request)
        throws TableStoreException, ClientException

以下樣本為 example_table 建立全量類型的通道 example_tunnel。

String tableName = "example_table";
String tunnelName = "example_tunnel";

CreateTunnelRequest request =
        new CreateTunnelRequest(
                tableName, tunnelName, TunnelType.BaseData);
CreateTunnelResponse response = tunnelClient.createTunnel(request);

System.out.println("TunnelId: " + response.getTunnelId());

參數說明

CreateTunnelRequest 包含以下參數：

名稱	類型	說明
tableName（必選）	`String`	資料表名稱。
tunnelName（必選）	`String`	通道名稱。
tunnelType（必選）	`TunnelType`	通道類型。`BaseData` 表示僅消費全量資料；`Stream` 表示僅消費增量資料；`BaseAndStream` 表示先消費全量資料，再消費增量資料。
streamTunnelConfig（可選）	`StreamTunnelConfig`	增量資料範圍配置，用於 `Stream` 和 `BaseAndStream` 類型的通道。如果設定此參數，時間範圍必須有效。
streamRecordOptions（可選）	`StreamRecordOptions`	增量記錄內容配置，用於 `Stream` 和 `BaseAndStream` 類型的通道。此參數需要使用 5.17.11 及以上版本的 Java SDK。

增量資料範圍

streamTunnelConfig 的類型為 StreamTunnelConfig，包含以下參數：

名稱	類型	說明
flag（可選）	`StartOffsetFlag`	未設定 `startOffset` 時的增量資料起點。`LATEST` 表示從通道建立時間開始，`EARLIEST` 表示從當前可讀取的最早增量日誌開始。預設值為 `LATEST`。
startOffset（可選）	`long`	增量資料的起始時間戳記。單位為毫秒，取值範圍為 [當前系統時間 - Stream 到期時間 + 5 分鐘，當前系統時間)。設定此參數後，`flag` 不生效。
endOffset（可選）	`long`	增量資料的結束時間戳記，單位為毫秒。同時設定起止時間時，此參數必須大於 `startOffset`。不設定時持續消費增量資料。

說明

Stream 到期時間是增量日誌的保留時間長度，最大值為 7 天。為資料表開啟 Stream 時可以設定該值，設定後不能修改。

增量記錄內容

streamRecordOptions 的類型為 StreamRecordOptions，包含以下參數：

名稱	類型	說明
getVersionGeneratorValue（可選）	`boolean`	增量記錄是否包含版本號碼產生器的值。預設值為 `false`。
getSysColumns（可選）	`boolean`	增量記錄是否包含系統列。預設值為 `false`。
getNewRowInfo（可選）	`boolean`	增量記錄是否包含最新行資訊。預設值為 `false`。
oldColumnsToGet（可選）	`StreamColumn`	原始行中需要返回的屬性列。
newColumnsToGet（可選）	`StreamColumn`	最新行中需要返回的屬性列。

增量記錄列

streamRecordOptions.oldColumnsToGet 和 streamRecordOptions.newColumnsToGet 的類型均為 StreamColumn，包含以下參數：

名稱	類型	說明
columnType（必選）	`StreamColumnType`	屬性列的選擇方式。`SPECIFIED_COLUMN` 表示指定屬性列，`INPUT_COLUMNS` 表示本次寫入或更新的屬性列，`ALL_COLUMNS` 表示所有屬性列。
columnNames（可選）	`List<String>`	`columnType` 為 `SPECIFIED_COLUMN` 時需要返回的屬性列名稱列表。

傳回值

CreateTunnelResponse 包含以下返回欄位：

欄位	類型	說明
tunnelId	`String`	建立的通道 ID，通過 `getTunnelId()` 擷取。消費通道資料時需要使用該 ID。

情境樣本

指定增量資料範圍

以下樣本建立增量類型的通道，並指定最近一小時內的增量資料範圍。

long endTime = System.currentTimeMillis() - 1_000L;
long startTime = endTime - 60 * 60 * 1000L;
StreamTunnelConfig streamConfig =
        new StreamTunnelConfig(startTime, endTime);

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table",
                "example_stream_tunnel",
                TunnelType.Stream);
request.setStreamTunnelConfig(streamConfig);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());

配置增量記錄內容

以下樣本建立增量類型的通道，並設定增量記錄返回版本號碼產生器的值、系統列、最新行資訊、原始行的 value 屬性列和最新行的所有屬性列。

StreamColumn oldColumns =
        new StreamColumn(StreamColumnType.SPECIFIED_COLUMN);
oldColumns.addColumnName("value");

StreamRecordOptions recordOptions = new StreamRecordOptions();
recordOptions.setGetVersionGeneratorValue(true);
recordOptions.setGetSysColumns(true);
recordOptions.setGetNewRowInfo(true);
recordOptions.setOldColumnsToGet(oldColumns);
recordOptions.setNewColumnsToGet(
        new StreamColumn(StreamColumnType.ALL_COLUMNS));

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table", "example_stream_tunnel", TunnelType.Stream);
request.setStreamRecordOptions(recordOptions);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());