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

本文介绍了如何并行而不是顺序执行多个查询?的处理方法，对大家解决问题具有一定的参考价值，需要的朋友们下面随着小编来一起学习吧！ 问题描述

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

I am querying all my 10 tables to get the user id from them and loading all the user id's into HashSet so that I can have unique user id.

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

As of now it is sequentially. We go to one table and extract all the user_id from it and load it in hash set and then second and third table and keep going.

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.

Is there any way to make this multithreaded so that for each table they get the data from my table in parallel? At the end, I need userList hashset which should have all the unique user id from all the 10 tables.

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

I am working with Cassandra database and connection is made only once so I don't need to create multiple connections.

推荐答案

如果您可以使用Java 8，则可以对表列表使用parallelStream进行此操作，并使用lambda扩展将表名放入每个表的唯一ID的对应列表中，然后将结果合并到单个哈希中.

If you're able to use Java 8, you could probably do this using parallelStream against a list of the tables, and use a lambda to expand the table name into the corresponding list of unique IDs per table, then join the results together into a single hash.

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

Without Java 8, I'd use Google Guava's listenable futures and an executor service something like this:

public static Set<String> fetchFromTable(int table) { String sql = "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.请参阅此处，以了解为什么后者更好.

The use of Executors and Futures is all core Java. The only thing Guava does is let me turn Futures into ListenableFutures. See here for a discussion of why the latter is better.

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

There are probably still ways to improve the parallelism of this approach, but if the bulk of your time is being spent in waiting for the DB to respond or in processing network traffic, then this approach may help.