将函数返回值中的对象添加到列表时的 Java 8 流异常处理

Question

我很难理解如何在使用 Java 8 流时处理异常。我想将对象添加到一个空列表中，并且每个对象都从一个函数返回，该函数可能会根据输入的内容引发异常。如果仅针对输入列表中的某些值引发异常，我希望它继续循环输入列表的其余部分。使用 for 循环似乎很容易做到：

  List<Item> itemList = new ArrayList<>();
  List<Input> inputs = getInputs(); //returns a list of inputs
  int exceptionCount = 0;

   // If an exception is thrown
   for (Input input : inputs){
        try {
           itemList.add(getItem(input));
        } catch (Exception e ) {
         // handle exception from getItem(input)
          exceptionCount = exceptionCount + 1;
        }  
   }

似乎可以使用Java流来实现这一点，但我不太确定如何处理

getItem()

函数可能抛出的异常。 这是我到目前为止所拥有的：

    final List<Item> itemList = new ArrayList<>();

     try {
         itemList = getInputs().stream()
                          .map(this::getItem)
                          .collect(Collectors.toList());
     } catch (Exception e ) {
         // handle exception from getItem(input)
          exceptionCount = exceptionCount + 1;
        } 

上面的方法显然行不通，因为一旦 getItem 抛出

one

异常，循环就不会继续，整个流只会抛出一个异常。有什么方法可以实现与 Java 8 流的基本 for 循环相同的实现吗？

Answer 1

您应该在

map

操作中捕获异常：

class ExceptionCounter {
    private int count = 0;

    void inc() { count++; }

    int getCount() { return count; }
}

ExceptionCounter counter = new ExceptionCounter();

List<Item> items = getInputs().stream()
    .map(input -> {
        try {
            return getItem(input);
        } catch (Exception e) {
            // handle exception here
            counter.inc();
        }
    })
    .collect(Collectors.toList());

虽然这对于顺序流可以按预期工作，但对于并行流则不起作用。即使流不是并行的，我们仍然需要

ExceptionCounter

持有者类，因为从流操作的参数（例如

map

）中引用的变量必须是有效的final。 （您可以使用包含一个元素的数组或

AtomicInteger

来代替 Holder 类）。

如果我们将

synchronized

添加到

inc

类的

ExceptionCounter

方法中，那么上面的解决方案将支持并行流。然而，

inc

方法的锁会出现大量的线程争用，从而失去并行化的优势。这（以及尝试不创建容易出错的代码）是不鼓励在流上使用“副作用”的原因。而计算异常的数量实际上是一个副作用。 对于这种特殊情况，如果使用自定义收集器，可以避免副作用：

class Result<R> {
    List<R> values = new ArrayList<>();
    int exceptionCount = 0;
    // TODO getters
}

static <T, R> Result<R> mappingToListCountingExceptions(Function<T, R> mapper) {
    class Acc {
        Result<R> result = new Result<>();

        void add(T t) {
            try {
                R res = mapper.apply(t);
                result.value.add(res);
            } catch (Exception e) {
                result.exceptionCount++;
            }
        }

        Acc merge(Acc another) {
            result.values.addAll(another.values);
            result.exceptionCount += another.exceptionCount;
        }
    }
    return Collector.of(Acc::new, Acc::add, Acc::merge, acc -> acc.result);
}

您可以按如下方式使用此自定义收集器：

Result<Item> items = getInputs().stream()
    .collect(mappingToListCountingExceptions(this::getItem));

现在由您来决定这种方法是否比传统的 

for

循环更好。

Answer 2

捕获地图中的异常

files.stream()
   .parallel()
   .map(file-> {
       try {
           return file.getInputStream();
       } catch (IOException e) {
           e.printStackTrace();
           return null;
       }
   })
   .forEach(inputStream -> carService.saveCars(inputStream));

---

提取函数参数以映射到其自己的方法中：

files.stream()
      .parallel()
      .map(file-> extractInputStream(file))
      .forEach(inputStream -> carService.saveCars(inputStream));

和

  private InputStream extractInputStream(MultipartFile file) {
        try {
            return file.getInputStream();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

---

创建另一个函数式接口，类似于Function，其apply方法确实声明了抛出异常：

@FunctionalInterface
interface FunctionWithException<T, R, E extends Exception> {
  R apply(T t) throws E;
}

和

private <T, R, E extends Exception>
    Function<T, R> wrapper(FunctionWithException<T, R, E> fun) {
        return arg -> {
                        try {
                            return fun.apply(arg);
                        } catch (Exception e) {
                            throw new RuntimeException(e);
                        }
        };
}

然后像这样使用它

files.stream()
     .parallel()
     .map(wrapper(file->file.getInputStream()))
     .forEach(inputStream -> carService.saveCars(inputStream));

---

但是如果你想高效地处理所有功能接口，我建议使用这个库

<dependency>
    <groupId>com.pivovarit</groupId>
    <artifactId>throwing-function</artifactId>
    <version>1.5.0</version>
</dependency>

本文中通过示例解释了所有内容

Java 8，如何处理流中的异常？

Answer 3