Java 8 流附加错误处理以供以后使用

Question

假设我有以下方法要重构

protected Stream<T> parseFile(File file, Consumer<File> cleanup) {
  try {
    return parser.parse(file); // returns a Stream<T>
  } catch (XmlParseException e) { // child of RuntimeException
    throw new CustomRuntimeException(e);
  } finally {
    if (file != null) {
      cleanup.accept(file);
    }
  }

  throw new IllegalStateException("Should not happen");
}

此方法的目的是充当代理，在包装异常CustomRuntimeException 中重新抛出流上的错误处理。因此，当我们稍后在流程中使用它时，我不必处处处理这些异常，而只需处理 CustomRuntimeException。

上游，我使用的方法如下

try {
  Stream<T> stream = parseFile(someFile);
  stream.map(t -> ...);
catch (CustomRuntimeException e) {
  // do some stuff
}

这就是 parser.parse 方法的样子

public Stream<T> parse() {
  // ValueIterator<T> implements Iterator<T>, AutoCloseable
  XmlRootParser.ValueIterator<T> valueIterator = new XmlRootParser.ValueIterator(this.nodeConverter, this.reader, this.nodeLocalName, this.nodeName);
  Stream<T> stream = StreamSupport.stream(Spliterators.spliteratorUnknownSize(valueIterator, 1040), false);
  stream.onClose(valueIterator::close);
  return stream;
}

我要处理的异常将由 ValueIterator.hasNext 方法抛出。这意味着它们不会在 Stream 创建时被抛出，而只会在 Stream 消费时抛出（在流上调用 foreach/map/count/collect/...）。

如何在parseFile 方法中很好地在我的流上附加错误处理，而不必使用流？有可能吗？

显然，此代码仅在 parser.parse 方法在返回之前消耗其流时才有效。这是反对使用流的。

Answer 1

提供迭代器逻辑的Stream的后端是Spliterator。

所以你可以像这样使用包装器Spliterator 来包装元素处理：

class Wrapper<T> implements Spliterator<T> {
    final Spliterator<T> source;
    public Wrapper(Spliterator<T> source) {
        this.source = source;
    }
    @Override
    public boolean tryAdvance(Consumer<? super T> action) {
        try {
            return source.tryAdvance(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override
    public void forEachRemaining(Consumer<? super T> action) {
        try {
            source.forEachRemaining(action);
        }
        catch(XmlParseException ex) {
            throw new CustomRuntimeException(ex);
        }
    }
    @Override public Spliterator<T> trySplit() {
        Spliterator<T> srcPrefix = source.trySplit();
        return srcPrefix == null? null: new Wrapper<>(srcPrefix);
    }
    @Override public long estimateSize() { return source.estimateSize(); }
    @Override public int characteristics() { return source.characteristics(); }
    @Override public Comparator<? super T> getComparator(){return source.getComparator();}
}

它保留原始Spliterator 的所有属性，并且只翻译迭代期间抛出的异常。

那么你就可以像这样使用它了

protected Stream<T> parseFile(File file) {
    Stream<T> s = parser.parse();
    return StreamSupport.stream(new Wrapper<>(s.spliterator()), s.isParallel())
                        .onClose(s::close);
}

调用者不要忘记正确关闭流：

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }

或

    ResultType result;
    try(Stream<T> s = parseFile(file)) {
        result = s.
         // other intermediate ops
         // terminal operation
    }
    finally {
        // other cleanup actions
    }

Answer 2

您可以使用辅助流初始化类来处理准备流的过程并在那里捕获任何异常。考虑以下示例：

import java.util.List;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class SafeInitializationStreamExample {

    public static void main(String[] args) {
        int sum = SafeInitializationStream.from(() -> Stream.of(1,2,3,4))
                .onInitializationError(t -> System.out.println(t.getMessage()))
                .mapToInt(it -> it)
                .sum();

        System.out.println(sum);

        List<Object> list = SafeInitializationStream.from(() -> parse("/tmp/test.log"))
                .onInitializationError(t -> System.out.println(t.getMessage()))
                .map(it -> it.toString())
                .collect(Collectors.toList());

        System.out.println(list);
    }

    private static <T> Stream<T> parse(String filename) {
        throw new RuntimeException("File does not exist!");
    }

    static class SafeInitializationStream<T> {
        private final Supplier<Stream<T>> streamSupplier;

        private SafeInitializationStream(Supplier<Stream<T>> streamSupplier) {
            this.streamSupplier = streamSupplier;
        }

        public static <T> SafeInitializationStream<T> from(Supplier<Stream<T>> streamSupplier) {
            return new SafeInitializationStream<>(streamSupplier);
        }

        public Stream<T> onInitializationError(Consumer<Throwable> onError) {
            try {
                return streamSupplier.get();
            } catch (Exception e) {
                onError.accept(e);
            }
            return Stream.empty();
        }
    }
}

在这个例子中，我们引入了SafeInitializationStream 类，它需要一个Supplier<Stream<T>>：

SafeInitializationStream.from(() -> Stream.of(1,2,3,4))

在这种情况下使用 Supplier<Stream<T>> 会使流初始化变得懒惰 - 直到我们调用 Supplier.get() 这个供应商的主体不会被执行。现在，当我们调用时：

.onInitializationError(t -> System.out.println(t.getMessage()))

我们执行供应商主体，捕获任何可能引发的异常，并通过将 Throwable 传递给作为参数传递给 onInitializationError 方法的 Consumer<Throwable> 来处理此异常。如果出现异常Stream.empty()，则返回（因此您可以安全地应用链中的所有其他转换。当没有异常时，返回供应商提供的Stream<T>。

如果您运行以下示例，您将进入控制台：

10
File does not exist!
[]

第一个流被使用，没有任何错误，sum 被正确返回。

第二个流在初始化过程中抛出异常，我们捕捉到它并打印到控制台，最后在消费流后返回空列表。

当然你可以在onInitializationError方法中使用Function<Throwable, Stream<T>>，如果你想指定在异常情况下返回什么Stream<T>。在这种情况下，我们假设在这种情况下总是返回Stream.empty()。希望对您有所帮助。