Tablestore:TableStoreReader を使用した同時データ読み取り

ステップ 1： Tablestore SDK をインストールする

Maven を使用して Java プロジェクトを管理する場合、pom.xml ファイルに次の依存関係を追加できます。

<dependency>
    <groupId>com.aliyun.openservices</groupId>
    <artifactId>tablestore</artifactId>
    <version>5.17.4</version>
</dependency>                 

詳細については、「Tablestore SDK をインストールする」をご参照ください。

ステップ 2： 初期化

TableStoreReader を初期化する前に、Tablestore Client 接続インスタンスを作成する必要があります。 TableStoreReader の構成パラメーターとコールバック関数をカスタマイズできます。 初期化のサンプルコードを以下に示します。

public static TableStoreReader createReader() {
        // Tablestore Client を初期化します
        client = new AsyncClient(endpoint, accessKeyId, accessKeySecret, instanceName);

        // TableStoreReader の構成
        TableStoreReaderConfig config = new TableStoreReaderConfig();

        // スレッドプール
        executorService = new ThreadPoolExecutor(4, 4, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

        // コールバック関数
        TableStoreCallback<PrimaryKeyWithTable, RowReadResult> callback = new TableStoreCallback<PrimaryKeyWithTable, RowReadResult>() {
            @Override
            public void onCompleted(PrimaryKeyWithTable primaryKeyWithTable, RowReadResult rowReadResult) {
                succeedRows.incrementAndGet();
                System.out.println(rowReadResult.getRowResult());
            }

            @Override
            public void onFailed(PrimaryKeyWithTable primaryKeyWithTable, Exception e) {
                failedRows.incrementAndGet();
                System.out.println("Failed Rows: " + primaryKeyWithTable.getTableName() + " | " + primaryKeyWithTable.getPrimaryKey() + " | " + e.getMessage()); // 失敗した行
            }
        };

        return new DefaultTableStoreReader(client, config, executorService, callback);
    }

// スキーマチェックを有効にするかどうかを設定します
config.setCheckTableMeta(false);
// バッファーキューサイズを設定します
config.setBufferSize(1024);
// 同時実行性を設定します
config.setConcurrency(10);
// バッチリクエストの最大行数を設定します
config.setMaxBatchRowsCount(100);
// 読み取るデータバージョンの数を設定します
config.setDefaultMaxVersions(1);
// フラッシュ間隔を設定します
config.setFlushInterval(10000);
// ログ間隔を設定します
config.setLogInterval(10000);
// バケット数を設定します
config.setBucketCount(4);

• コールバック関数が不要な場合は、初期化メソッドでコールバック関数パラメーターを null に設定できます。

  return new DefaultTableStoreReader(client, config, executorService, null);

ステップ 3： データをクエリする

1. TableStoreReader を使用してデータをクエリする前に、行データのプライマリキー情報をバッファーに追加する必要があります。

   PrimaryKey primaryKey = PrimaryKeyBuilder.createPrimaryKeyBuilder()
           .addPrimaryKeyColumn("id", PrimaryKeyValue.fromString("row1"))
           .build();
   tableStoreReader.addPrimaryKey("test_table", primaryKey);

   • クエリ後に行データを取得する必要がある場合は、addPrimaryKeyWithFuture メソッドを使用できます。

     Future<ReaderResult> readerResult = tableStoreReader.addPrimaryKeyWithFuture("test_table", primaryKey);

   • 最大バージョン番号、データバージョン範囲、フィルターなど、クエリパラメーターを指定することもできます。

     RowQueryCriteria rowQueryCriteria = new RowQueryCriteria("test_version");
     // 読み取る最大バージョン数を設定します
     rowQueryCriteria.setMaxVersions(1);
     // 読み取るデータバージョン範囲を設定します
     rowQueryCriteria.setTimeRange(new TimeRange(System.currentTimeMillis() - 86400*1000, System.currentTimeMillis()));
     // 返す属性列を設定します
     rowQueryCriteria.addColumnsToGet("col1");
     // フィルター条件を設定します
     SingleColumnValueFilter singleColumnValueFilter = new SingleColumnValueFilter("col1", SingleColumnValueFilter.CompareOperator.EQUAL, ColumnValue.fromString("val1"));
     rowQueryCriteria.setFilter(singleColumnValueFilter);
     // クエリ条件を追加します
     tableStoreReader.setRowQueryCriteria(rowQueryCriteria);

2. プライマリキー情報がバッファーに追加されると、TableStoreReader は設定された自動送信時間間隔（デフォルトは 10 秒）に従って、バッファー内のデータをクエリのために Tablestore に送信します。 バッファーデータを手動で送信することもできます。

   • 同期送信

     tableStoreReader.flush();

   • 非同期送信

     tableStoreReader.send();

ステップ 4： リソースを無効にする

データクエリの完了後、他に操作が必要ない場合は、業務システムの動作に影響を与えることなくリソースを無効にできます。

tableStoreReader.close();
client.shutdown();
executorService.shutdown();