Parallel.ForEach 中try catch 对异常信息保存

在使用 Parallel.ForEach 时，如果在并行执行的过程中发生了异常，可以使用 try/catch 块来捕获并处理异常。如果希望在 catch 块中记录异常信息，可以使用以下代码：

List<Exception> exceptions = new List<Exception>();

Parallel.ForEach(items, item =>
{
    try
    {
        // 执行操作
    }
    catch (Exception ex)
    {
        // 记录异常信息
        exceptions.Add(ex);
    }
});

if (exceptions.Count > 0)
{
    // 处理异常信息
}

在上面的示例中，我们创建了一个 List<Exception> 类型的变量 exceptions 来保存捕获到的异常信息。在 Parallel.ForEach 中的 try/catch 块中，如果发生了异常，就将异常信息添加到 exceptions 列表中。在并行操作完成后，我们可以检查 exceptions 列表是否为空，如果不为空，就可以处理异常信息。

相关问题

parallelStream并行流，解决main线程异常信息包括了fork进程的异常类型“java.lang.RuntimeExcetion”及异常信息“1异常了”。

如果您使用`parallelStream`方法处理流时，希望在主线程中捕获子线程中发生的异常，可以使用`CompletableFuture`的`join`方法。`join`方法会等待并行流中的所有任务完成，并返回流的结果。如果在任务执行过程中发生了未捕获的异常，则`join`方法会将异常重新抛出，以便在主线程中捕获和处理它。

下面是一个例子，演示如何使用`join`方法捕获并行流中的异常：

List<Integer> numbers = Arrays.asList(1, 2, 3, 4, 5);

try {
    numbers.parallelStream()
           .map(n -> {
                if (n == 1) {
                    throw new RuntimeException("1异常了");
                }
                return n;
           })
           .forEach(System.out::println);
} catch (Exception e) {
    System.out.println("Caught exception: " + e.getMessage());
}

// Using join method to catch exceptions
try {
    numbers.parallelStream()
           .map(n -> CompletableFuture.supplyAsync(() -> {
                if (n == 1) {
                    throw new RuntimeException("1异常了");
                }
                return n;
           }))
           .map(CompletableFuture::join)
           .forEach(System.out::println);
} catch (Exception e) {
    System.out.println("Caught exception: " + e.getMessage());
}

在这个例子中，我们首先使用`parallelStream`方法处理流，并在`map`操作中抛出一个`RuntimeException`，以模拟子线程中发生的异常。我们在`try-catch`块中捕获这个异常，并打印出错误消息。

然后，我们使用`CompletableFuture.supplyAsync`方法将`map`操作包装成一个`CompletableFuture`，以便在子线程中执行。我们使用`join`方法等待所有`CompletableFuture`完成，并将结果收集到一个新的流中。如果在执行过程中发生了异常，则`join`方法会将异常重新抛出，以便在主线程中捕获和处理它。

serverWatches.forEach( sw -> { String serverId = sw.getId(); String ip = sw.getIp(); try { saveSystemMonitor(serverId, ip, time, plmxs_url); } catch (Exception e) { log.error("保存cpu使用率|内存使用率失败:{}", e); } try { saveDiskIoDetail(serverId, ip, time, plmxs_url); } catch (Exception e) { log.error("保存磁盘IO失败:{}", e); } try { saveNetIoReceive(serverId, ip, time, plmxs_url); } catch (Exception e) { log.error("保存网络IO-receive失败:{}", e); } try { saveNetIoTransmit(serverId, ip, time, plmxs_url); } catch (Exception e) { log.error("保存网络IO-transmit失败:{}", e); } } );

可以将每个保存操作都放入一个线程中进行执行，这样可以提高并发处理能力，代码如下所示：

serverWatches.parallelStream().forEach(sw -> {
    String serverId = sw.getId();
    String ip = sw.getIp();
    ExecutorService executor = Executors.newFixedThreadPool(4);
    executor.execute(() -> {
        try {
            saveSystemMonitor(serverId, ip, time, plmxs_url);
        } catch (Exception e) {
            log.error("保存cpu使用率|内存使用率失败:{}", e);
        }
    });
    executor.execute(() -> {
        try {
            saveDiskIoDetail(serverId, ip, time, plmxs_url);
        } catch (Exception e) {
            log.error("保存磁盘IO失败:{}", e);
        }
    });
    executor.execute(() -> {
        try {
            saveNetIoReceive(serverId, ip, time, plmxs_url);
        } catch (Exception e) {
            log.error("保存网络IO-receive失败:{}", e);
        }
    });
    executor.execute(() -> {
        try {
            saveNetIoTransmit(serverId, ip, time, plmxs_url);
        } catch (Exception e) {
            log.error("保存网络IO-transmit失败:{}", e);
        }
    });
    executor.shutdown();
});

在上面的代码中，我们使用了 Java 自带的线程池进行多线程处理，其中 `newFixedThreadPool(4)` 表示创建一个固定大小为 4 的线程池，每个保存操作都会被分配到其中一个线程中进行执行。最后，我们需要在代码结束前调用 `executor.shutdown()` 来关闭线程池。