使用 ResultSetFuture 修改集合

Question

我尝试使用实现异步查询 java-driver-async-queries。我正在修改 FutureCallback 中的列表，但似乎它不起作用 -

List<Product> products = new ArrayList<Product>();

for (// iterating over a Map) {
    key = entry.getKey();
    String query = "SELECT id,desc,category FROM products where id=?";
    ResultSetFuture future = session.executeAsync(query, key);
    Futures.addCallback(future,
        new FutureCallback<ResultSet>() {
            @Override public void onSuccess(ResultSet result) {
                Row row = result.one();
                if (row != null) {
                    Product product = new Product();
                    product.setId(row.getString("id"));
                    product.setDesc(row.getString("desc"));
                    product.setCategory(row.getString("category"));

                    products.add(product);
                }
            }

            @Override public void onFailure(Throwable t) {
                // log error
            }
        },
        MoreExecutors.sameThreadExecutor()
    );
}

System.out.println("Product List : " + products); // Not printing correct values. Sometimes print blank

还有其他方法吗？

根据我实施的 Mikhail Baksheev 回答，现在得到了正确的结果。 只是一个转折。我需要实现一些额外的逻辑。我想知道我是否可以使用 List<MyClass> 而不是 List<ResultSetFuture> 和 MyClass 作为 -

public class MyClass {

    private Integer         productCount;
    private Integer         stockCount;
    private ResultSetFuture result;
}

然后在迭代时将 FutureList 设置为 -

ResultSetFuture result = session.executeAsync(query, key.get());
MyClass allResult = new MyClass();
allResult.setInCount(inCount);
allResult.setResult(result);
allResult.setSohCount(values.size() - inCount);

futuresList.add(allResult);

Answer 1

正如@RussS 提到的，代码不会等待所有期货都完成。

有很多方法可以同步异步代码。例如，使用CountDownLatch:

编辑： 也请使用单独的线程进行回调，并使用并发收集产品。

ConcurrentLinkedQueue<Product> products = new ConcurrentLinkedQueue<Product>();
final Executor callbackExecutor = Executors.newSingleThreadExecutor();
final CountDownLatch doneSignal = new CountDownLatch(/*the Map size*/);
for (// iterating over a Map) {
    key = entry.getKey();
    String query = "SELECT id,desc,category FROM products where id=?";
    ResultSetFuture future = session.executeAsync(query, key);
    Futures.addCallback(future,
        new FutureCallback<ResultSet>() {
            @Override public void onSuccess(ResultSet result) {
                Row row = result.one();
                if (row != null) {
                    Product product = new Product();
                    product.setId(row.getString("id"));
                    product.setDesc(row.getString("desc"));
                    product.setCategory(row.getString("category"));

                    products.add(product);
                }
                doneSignal.countDown();

            }

            @Override public void onFailure(Throwable t) {
                // log error
                doneSignal.countDown();
            }
        },
        callbackExecutor
    );
}

doneSignal.await();           // wait for all async requests to finish
System.out.println("Product List : " + products); 

另一种方法是收集列表中的所有期货，并使用 guava 的 Futures.allAsList 将所有结果作为单个未来等待，例如：

List<ResultSetFuture> futuresList = new ArrayList<>( /*Map size*/);
        for (/* iterating over a Map*/) {
            key = entry.getKey();
            String query = "SELECT id,desc,category FROM products where id=?";
            futuresList.add( session.executeAsync( query, key ) );
        }

        ListenableFuture<List<ResultSet>> allFuturesResult = Futures.allAsList( futuresList );
        List<Product> products = new ArrayList<>();
        try {
            final List<ResultSet> resultSets = allFuturesResult.get();
            for ( ResultSet rs : resultSets ) {
                if ( null != rs ) {
                    Row row = rs.one();
                    if (row != null) {
                        Product product = new Product();
                        product.setId(row.getString("id"));
                        product.setDesc(row.getString("desc"));
                        product.setCategory(row.getString("category"));

                        products.add(product);
                    }
                }
            }
        } catch ( InterruptedException | ExecutionException e ) {
            System.out.println(e);
        }
        System.out.println("Product List : " + products);

编辑 2

我想知道我是否可以使用 List 而不是 List 和 MyClass as

技术上是的，但是在这种情况下你不能在Futures.allAsList 中传递List<MyClass> 或者MyClass 应该实现ListenableFuture 接口