并发执行Java 8

Question

我刚开始使用java进行并发编程，并尝试异步启动一些Callables。但是代码似乎阻止了我的程序流，其中Callables被赋予了executorService es.invokeAll(tasks)。

public void checkSensorConnections(boolean fireEvent) {
    List<Callable<Void>> tasks = new ArrayList<>();

    getSensors().forEach(sensor -> {
        tasks.add(writerService.openWriteConnection(sensor));
        tasks.add(readerService.openReadConnection(sensor));
    });

    try {
        LOG.info("Submmitting tasks");

        ExecutorService es = Executors.newWorkStealingPool();
        es.invokeAll(tasks);

        LOG.info("Tasks submitted");
    } catch (InterruptedException e) {
        LOG.error("could not open sensor-connections", e);
        error(MeasurmentScrewMinerError.OPEN_CONNECTION_ERROR);
    }
}

我有一些日志语句控制程序的流程。正如您所看到的，执行等待两个任务的执行。

2017-01-19 16:06:06:06,474 INFO de.cgh.screwminer.service.measurement.MeasurementService (MeasurementService.java:127) -转交任务 177-01-19 16:06:08,477错误池-2-线程-2 de.cgh.screwminer.service.measurement.SensorReadService (SensorReadService.java:68) -传感器读连接无法打开java.net.SocketTimeoutException:接收超时。 177-01-19 16:06:08,477错误池-2-线程-4 de.cgh.screwminer.service.measurement.SensorReadService (SensorReadService.java:68) -传感器卡夫读取-连接无法打开java.net.SocketTimeoutException:接收超时. 2017-01-19 16:06:08,482 INFO main de.cgh.screwminer.service.measurement.MeasurementService (MeasurementService.java:132) -任务提交

Answer 1

来自invokeAll的Javadoc：

Returns:
   a list of Futures representing the tasks, in the same sequential order as produced by the iterator for the given task list, each of which has completed

因此，是的，invokeAll任务已经完成。

您所能做的就是将执行器保存在类中，并在forEach()中提交每个任务，这将执行相同的任务。然后，你会得到一个期货的列表，你应该检查错误。

你可以这样做：

getSensors().forEach(s -> {
    CompletableFuture<Void> cf = (s -> writerService.openWriteConnection(s)).exceptionally(ex -> errorhandling)
    exec.submit(cf)
});

CompletableFuture是一个Java8特性，您可以很好地控制错误，因为您不需要询问“期货”是否成功完成这些错误(这通常会导致意外的非日志错误)