我正在努力优化我的Future的管理技术。
假设我们有这种典型的处理场景：我运行一个查询以从数据库中获取一些记录：
SELECT * FROM mytable WHERE mycondition;
该查询返回很多我需要处理的行，例如：
while (recordset have more results) {
MyRow row = recordset.getNextRow(); // Get the next row
processRow(row); // Process the row
}
现在假设所有行彼此独立，并且函数processRow速度很慢，因为它在 C* 集群上执行一些硬处理和查询：
void processRow(MyRow row) {
// Fetch some useful data from the DB
int metadataid = row.getMetadataID();
Metadata metadata = getMetadataFromCassandra(metadataid);
// .... perform more processing on the row .....
// Store the processing result in the DB
ProcessingResult result = ....;
insertProcessingResultIntoCassandra(result);
}
像这样的串行方法预计性能很差，因此并行执行是有争议的。
考虑到这个基本的处理结构，以下是我对算法进行的一些转换，以获得主要的速度提升。
第 1 步：并行化行处理
这非常简单。我创建了一个Executor得到的jobs并行完成。然后我等待所有工作完成。代码如下：
ThreadPoolExecutor executor = (ThreadPoolExecutor)Executors.newCachedThreadPool();
int failedJobs = 0;
ArrayList<Future<Boolean>> futures = new ArrayList<>();
while (recordset have more results) {
final MyRow row = recordset.getNextRow(); // Get the next row
// Create the async job and send it to the executor
Callable<Boolean> c = new Callable<Boolean>() {
@Override
public Boolean call() {
try {
processRow(row);
} catch (Exception e) {
return false; // Job failed
}
return true; // Job is OK
}
};
futures.add(executor.submit(c));
}
// All jobs submitted. Wait for the completion.
while (futures.size() > 0) {
Future<Boolean> future = futures.remove(0);
Boolean result = false;
try {
result = future.get();
} catch (Exception e) {
e.printStackTrace();
}
failedJobs += (result ? 0 : 1);
}
第 2 步：限制并发行数
到目前为止一切顺利，除非我的数量很少jobs这预计会因内存不足错误而失败，因为执行器由未绑定队列支持，并且主循环将一路提交作业。我可以通过控制并发提交作业的最大数量来解决这个问题：
final const int MAX_JOBS = 1000;
while (recordset have more results) {
....
futures.add(executor.submit(c));
while (futures.size() >= MAX_JOBS) {
Future<Boolean> future = futures.remove(0);
Boolean result = false;
try {
result = future.get();
} catch (Exception e) {
e.printStackTrace();
}
failedJobs += (result ? 0 : 1);
}
}
简单地说，如果达到某个阈值（在本例中为 1000），我会等待列表中的第一项作业完成。这很有效，而且加速效果很好。
第 3 步：并行化单行处理
这是我想获得一些帮助的步骤。我预计1000由于 IO 缓慢，作业将在队列中快速积累。也就是说，我希望 JVM 能够启动1000线程来容纳所有的工作。现在，当你只有 8 核机器时，1000 个线程通常会减慢一切，我认为通过更多调整并行性，这个数字可以降低。
目前，getMetadataFromCassandra函数是一个包装器session.executeAsync，但管理重试：
public static ResultSet getMetadataFromCassandra(...) {
int retries = 0;
// Loop here
while (retries < MAX_RETRIES) {
// Execute the query
ResultSetFuture future = session.executeAsync(statement);
try {
// Try to get the result
return future.get(1000 * (int)Math.pow(2, retries), TimeUnit.MILLISECONDS);
} catch (Exception e) {
// Ooops. An error occurred. Cancel the future and schedule it again
future.cancel(true);
if (retries == MAX_RETRIES) {
e.printStackTrace();
String stackTrace = Throwables.getStackTraceAsString(e);
logToFile("Failed to execute query. Stack trace: " + stackTrace);
}
retries++;
}
}
return null;
}
正如你所看到的，这是一个阻塞函数，因为我.get() on the ResultSetFuture。也就是说，这个调用会阻塞每个等待IO的线程。所以我采用了异步方法，但我觉得我浪费了很多硬件资源。
QUESTION
在我看来，当.executeAsync结果可用（或发生超时），“释放”线程并允许同一线程执行其他操作。
简而言之，在我看来，我需要改变顺序的的结构processRow into a pipeline：查询以异步方式执行，当结果可用时，将执行处理的其余部分。当然，我希望主循环等待整个流水线式的完成的过程，而不仅仅是第一部分。
换句话说，主循环提交一个作业（我们称之为jobJob）我得到一个Future（我们打电话jobFuture）我可以.get()等待其完成。然而，jobJob触发“查询”子作业（我们称之为queryJob), and queryJob是异步提交的，所以我得到另一个Future（我们称之为queryFuture）应该用来触发“进程”子作业（让我们调用processJob）。此时，我只是在嵌套Futures并在完成之前阻止链深处Future代表jobJob，这意味着我又回到了原点！！！
在我采取困难的路线并将这种管道实现为有限状态机之前，我查看了：
-
ForkJoinPool执行者类
-
ListenableFuture来自Guava library
-
CompletableFuture class
它们似乎都不能满足我对这个过程进行流水线处理的要求，或者可能我没有找到关于如何执行如此明显的简单任务的明确解释。谁能简单地告诉我这个话题？
非常感谢任何帮助。
