如何并行而不是顺序执行多个查询？

Question

我要查询所有10个表以从中获取用户ID，并将所有用户ID加载到HashSet中，以便可以拥有唯一的用户ID。

截至目前，它是按顺序进行的。我们转到一个表，并从中提取所有user_id并将其加载到哈希集中，然后将其存储在第二个和第三个表中并继续运行。

    private Set<String> getRandomUsers() {
        Set<String> userList = new HashSet<String>();

        // is there any way to make this parallel?
        for (int table = 0; table < 10; table++) {
            String sql = "select * from testkeyspace.test_table_" + table + ";";

            try {
                SimpleStatement query = new SimpleStatement(sql);
                query.setConsistencyLevel(ConsistencyLevel.QUORUM);
                ResultSet res = session.execute(query);

                Iterator<Row> rows = res.iterator();
                while (rows.hasNext()) {
                    Row r = rows.next();

                    String user_id = r.getString("user_id");
                    userList.add(user_id);
                }
            } catch (Exception e) {
                System.out.println("error= " + ExceptionUtils.getStackTrace(e));
            }
        }

        return userList;
    }

有什么方法可以使这个多线程的，以便他们为每个表并行地从我的表中获取数据？最后，我需要userList哈希集，该哈希集应具有所有10个表中的所有唯一用户ID。

我正在使用Cassandra数据库，并且连接仅建立一次，因此不需要创建多个连接。

Answer 1

如果您能够使用Java
8，则可以parallelStream对表列表进行操作，并使用lambda将表名扩展为每个表的唯一ID对应列表，然后将结果合并为一个表哈希。

没有Java 8，我将使用Google Guava的可监听期货和类似以下内容的执行服务：

public static Set<String> fetchFromTable(int table) {
    String sql = "html" target="_blank">select * from testkeyspace.test_table_" + table + ";";
    Set<String> result = new HashSet<String>();
    // populate result with your SQL statements
    // ...
    return result;
}

public static Set<String> fetchFromAllTables() throws InterruptedException, ExecutionException {
    // Create a ListeningExecutorService (Guava) by wrapping a 
    // normal ExecutorService (Java) 
    ListeningExecutorService executor = 
            MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());

    List<ListenableFuture<Set<String>>> list = 
            new ArrayList<ListenableFuture<Set<String>>>(); 
    // For each table, create an independent thread that will 
    // query just that table and return a set of user IDs from it
    for (int i = 0; i < 10; i++) {
        final int table = i;
        ListenableFuture<Set<String>> future = executor.submit(new Callable<Set<String>>() {
            public Set<String> call() throws Exception {
                return fetchFromTable(table);
            }
        });
        // Add the future to the list
        list.add(future);
    }
    // We want to know when ALL the threads have completed, 
    // so we use a Guava function to turn a list of ListenableFutures
    // into a single ListenableFuture
    ListenableFuture<List<Set<String>>> combinedFutures = Futures.allAsList(list);

    // The get on the combined ListenableFuture will now block until 
    // ALL the individual threads have completed work.
    List<Set<String>> tableSets = combinedFutures.get();

    // Now all we have to do is combine the individual sets into a
    // single result
    Set<String> userList = new HashSet<String>();
    for (Set<String> tableSet: tableSets) {
        userList.addAll(tableSet);
    }

    return userList;
}

Executors和Futures的使用都是Java的核心。番石榴唯一要做的就是让我将Future变成ListenableFutures。请参阅此处以讨论为何后者更好。

可能仍有改善这种方法并行性的方法，但是如果您花费大量时间等待数据库响应或处理网络流量，则此方法可能会有所帮助。