Dynamo DB中的分页结果具有完全的未来

Question

我正在为给定的主键查询Dynamo DB。主键由两个UUID字段（fieldUUID1、fieldUUID2）组成。对于上面的主键组合和值列表，我有很多查询要执行。为此，我使用异步CompletableFuture和ExecutorService，线程池大小为4。

在所有查询返回结果之后，这是CompletableFuture

如果某些查询导致结果分页，即返回lastEvaluatedKey，则我无法知道哪个查询请求返回了此结果。

如果我做了一个. get（）调用，而我收到了完全未来，这将是一个阻塞操作，这违背了使用异步的目的。有没有办法让我处理这种情况？

例如：

我可以尝试使用Compose方法，但是当lastEvaluatedKey不存在时，我如何知道需要在什么时候停止。

for (final QueryRequest queryRequest : queryRequests) {
    final CompletableFuture<QueryResult> futureResult =
        CompletableFuture.supplyAsync(() ->
            dynamoDBClient.query(queryRequest), executorService));

    if (futureResult == null) {
        continue;
    }

    futures.add(futureResult);
}

// Wait for completion of all of the Futures provided
final CompletableFuture<Void> allfuture = CompletableFuture
    .allOf(futures.toArray(new CompletableFuture[futures.size()]));

// The return type of the CompletableFuture.allOf() is a
// CompletableFuture<Void>. The limitation of this method is that it does not
// return the combined results of all Futures. Instead we have to manually get
// results from Futures. CompletableFuture.join() method and Java 8 Streams API
// makes it simple:
final CompletableFuture<List<QueryResult>> allFutureList = allfuture.thenApply(val -> {
    return futures.stream().map(f -> f.join()).collect(Collectors.toList());
});


final List<QueryOutcome> completableResults = new ArrayList<>();
try {
    try {
        // at this point all the Futures should be done, because we already executed
        // CompletableFuture.allOf method.
        final List<QueryResult> returnedResult = allFutureList.get();
        for (final QueryResult queryResult : returnedResult) {
            if (MapUtils.isNotEmpty(queryResult.getLastEvaluatedKey()) {
                // how to get hold of original request  and include last evaluated key ?
            }
        }
    } finally {

    }
} finally {

}

我可以依赖. get（）方法，但它将是一个阻塞调用。




Answer 1

满足您需求的快速解决方案是更改您的未来列表。而不是让它存储CompletableFuture

然后，一旦allfuture完成，您就可以迭代futures并访问请求和结果。

然而，这里有一个更深层次的问题。一旦您访问了原始查询请求，您打算做什么？我猜您希望发出一个后续请求，并将exclusiveStartKey设置为响应的lastEvaluatedKey所包含的任何值。这意味着您将等待所有原始查询完成，然后才发布下一批查询。这是低效的：如果查询返回了lastEvaluatedKey，您希望尽快发出后续查询。

为了实现这一点，我建议你引入一个新方法，它只需要一个QueryRequest对象，并返回一个CompletableFuture

• 用给定的请求发出查询
• 一旦结果到达，检查它。如果它的lastEvalue atedKey是空的，返回它作为方法的结果
• 否则，更新request.exclusiveStartKey并返回第一步。

是的，使用CompletableFutures（与阻塞代码相比）要做到这一点有点困难，但完全可行。

一旦你有了这个方法，你的代码需要对query请求中的每个请求调用这个方法一次，把返回的CompletableFuture放在一个列表中，并在上面做一个CompletableFuture.allOf（）那个名单。一旦所有的未来完成，你可以只使用结果-不需要做问题后续查询。