java - 如何并行处理 Iterables.partition(...) 结果以与 BatchGetItem API 一起使用？

Question

我正在尝试调用 BatchGetItem 从 DynamoDB 中检索项目。作为输入，我们可以获得最多 1000 个键（或少至 1 个键）的列表。这些键与我们的 DynamoDB 表的 hashKey 一致。

由于 BatchGetItem API 每次调用最多只能接收 100 个项目，因此我尝试将请求拆分为每批仅 100 个项目，并行进行调用，然后再次将结果合并到一个 Set 中。

对于那些不熟悉 DynamoDB 仍然可以就极其精简的版本（第一个示例）提供建议的人，我将不胜感激！否则，请参阅下面的第二个更准确的示例。

第一个例子 - 极度精简

public Set<SomeResultType> retrieveSomething(Set<String> someSet) {
   ImmutableSet.Builder<SomeResultType> resultBuilder = ImmutableSet.builder();

   // FIXME - how to parallelize? 
   for (List<Map<String, String>> batch : Iterables.partition(someSet, 100)) {
       result = callSomeLongRunningAPI(batch);
       resultBuilder.addAll(result.getItems());
   }
   return resultBuilder.build();
}

第二个例子 - 更接近我的实际问题 - 下面是我目前正在做的一个精简的虚拟版本（因此，请原谅格式/样式问题）。它目前可以工作并获取所有项目，但我无法弄清楚如何让批次（请参阅 FIXME）并行执行并最终成为一个集合。由于性能在我正在尝试构建的系统中非常重要，因此任何提示将不胜感激，以帮助此代码更高效！

public Set<SomeResultType> retrieveSomething(Set<String> someIds) {
    if (someIds.isEmpty()) {
        // handle this here
    }

    Collection<Map<String, AttributeValue>> keyAttributes = someIds.stream()
            .map(id -> ImmutableMap.<String, AttributeValue>builder()
                    .put(tableName, new AttributeValue().withS(id)).build())
            .collect(ImmutableList.toImmutableList());

    ImmutableSet.Builder<SomeResultType> resultBuilder = ImmutableSet.builder();
    Map<String, KeysAndAttributes> itemsToProcess;
    BatchGetItemResult result;

    // FIXME - make parallel?
    for (List<Map<String, AttributeValue>> batch : Iterables.partition(keyAttributes, 100)) {
        KeysAndAttributes keysAndAttributes = new KeysAndAttributes()
                .withKeys(batch)
                .withAttributesToGet(...// some attribute names);

        itemsToProcess = ImmutableMap.of(tableName, keysAndAttributes);
        result = this.dynamoDB.batchGetItem(itemsToProcess);

        resultBuilder.addAll(extractItemsFromResults(tableName, result));
    }

    return resultBuilder.build());
}

非常感谢您对超级精简案例或第二个示例的帮助！谢谢！

标签： javajava-8amazon-dynamodb