java - 如何仍然使用 Future.get 管理并行化？

Question

我想并行运行线程，并在任何线程运行时失败时处理异常。但是，当我使用 future.get 方法时，我无法维持并行执行。如何解决这个问题？

for (int i = 0; i < threads; i++) {
    final Future<Void> future = executor.submit(new ReaderDirectWithPartition_JDBC(outputPath + i, partitionedQuery, propparameters, props));

    try {
        future.get();
    } catch (final CancellationException ce) {
        log.error(ce.getMessage());
        executor.shutdownNow();
    } catch (final ExecutionException ee) {
        log.error(ee.getMessage());
        executor.shutdownNow();
    } catch (final InterruptedException ie) {
        Thread.currentThread().interrupt(); // ignore/reset
        log.error(ie.getMessage());
        executor.shutdownNow();
    }
}

Answer 1

首先在一个循环中将所有操作提交到线程池 - 收集列表或数组中的 Future 元素。并在单独的循环中调用 Future.get()。

<小时/>

final MultiTaskStatus sharedStatus = new MultiTaskStatus();
final List<Future<Void>> pendingsResults = new ArrayList<Future<Void>>(threads);
// submit all the jobs to be executed by the thread pool (executor)
for (int i = 0; i < threads; i++) {
    // avoid submitting more tasks, if one already failed
    if (sharedStatus.isFailed()) {
        break;
    }
    final ReaderDirectWithPartition_JDBC job;
    // hand over the MultiTaskStatus to be set as failed on error
    job = new ReaderDirectWithPartition_JDBC(outputPath + i,
            partitionedQuery, propparameters, props, sharedStatus);
    pendingResults.add(executor.submit(job));
}
if (sharedStatus.isFailed()) {
    // maybe do something special, if we already know something went wrong?
}
try {
    // wait for all scheduled jobs to be done
    for (final Future<Void> pendingJob : pendingResults) {
        pendingJob.get();
    }
} catch (final CancellationException ce) {
    log.error(ce.getMessage());
} catch (final ExecutionException ee) {
    log.error(ee.getMessage());
} catch (final InterruptedException ie) {
    log.error(ie.getMessage());
} finally {
    executor.shutdownNow();
}

带有单独的状态指示器

class MultiTaskStatus {
    private boolean failed = false;
    public synchronized void setFailed(boolean taskFailed) {
        this.failed = this.failed || taskFailed;
    }
    public synchronized boolean isFailed() {
        return this.failed;
    }
}

也许可以考虑使用 this answer 中提到的 CompletionService .