java - 流式传输时重新抛出异常

Question

我在“访问”流时重新抛出流抛出的异常时遇到问题。

例如，如果我有一个抛出ExceptionA的流:

Stream<String> stream = Stream.of("dummy").map(d -> {throw new ExceptionA();});

try {
   stream.collect(Collectors.toList());
} catch (ExceptionA e) {}

我想要实现的是从 stream 创建新的 stream2 而不消耗 stream 这将抛出 ExceptionB 收集时

try {
   stream2.collect(Collectors.toList());
} catch (ExceptionB e) {}

显然

Iterator<String> newIt = createRethrowingIterator(stream.iterator());
Stream<String> stream2 = StreamSupport.stream(Spliterators.spliteratorUnknownSize(newIt, Spliterator.NONNULL), false)

其中createRethroingIterator包装原始迭代器并返回新的迭代器，该迭代器实际上将ExceptionA重新抛出ExceptionB

不是我想要的，因为 stream.iterator() 是终端运算符，即它将消耗流，如果流非常大，这可能会导致内存问题

Answer 1

使用 Spliterator 而不是 Iterator 可以更好地解决此任务。它简化了逻辑，因为您只需通过委托(delegate)源的 tryAdvance 方法来实现一个方法 tryAdvance。

它还通过将方法 characteristics() 和 estimateSize() 委托(delegate)给源来提供性能改进的机会，因为异常转换功能不会更改它们。您还可以通过委托(delegate)给源来实现 trySplit 来获得良好的并行支持。您只需像第一个 Spliterator 一样包装结果即可:

public class Rethrowing<T,E extends Throwable> implements Spliterator<T> {
    public static <E extends Throwable, T> Stream<T> translateExceptions(
        Stream<T> source, Class<E> catchType,
        Function<? super E, ? extends RuntimeException> translator) {

        return StreamSupport.stream(new Rethrowing<>(
            source.spliterator(), catchType, translator), source.isParallel());
    }
    private final Spliterator<T> source;
    private final Class<E> catchType;
    private final Function<? super E, ? extends RuntimeException> translator;

    public Rethrowing(Spliterator<T> sp, Class<E> catchType,
            Function<? super E, ? extends RuntimeException> translator) {
        this.source = sp;
        this.catchType = catchType;
        this.translator = translator;
    }
    @Override public boolean tryAdvance(Consumer<? super T> action) {
        try { return source.tryAdvance(action); }
        catch(Throwable t) {
            if(catchType.isInstance(t))
                throw translator.apply(catchType.cast(t));
            else throw t;
        }
    }
    @Override public int characteristics() {
        return source.characteristics();
    }
    @Override public long estimateSize() {
        return source.estimateSize();
    }
    @Override public Spliterator<T> trySplit() {
        Spliterator<T> split = source.trySplit();
        return split==null? null: new Rethrowing<>(split, catchType, translator);
    }
}

您可以像这样使用这个实用程序类

class ExceptionA extends IllegalStateException {
    public ExceptionA(String s) {
        super(s);
    }
}
class ExceptionB extends IllegalStateException {
    public ExceptionB(Throwable cause) {
        super(cause);
    }
}
Rethrowing.translateExceptions(
    Stream.of("foo", "bar", "baz", "", "extra")
        .peek(s -> System.err.println("consuming \""+s+'"'))
        .map(s -> { if(s.isEmpty()) throw new ExceptionA("empty"); return s; }),
    ExceptionA.class, ExceptionB::new)
        .forEach(s -> System.err.println("terminal operation on "+s));

获取

consuming "foo"
terminal operation on foo
consuming "bar"
terminal operation on bar
consuming "baz"
terminal operation on baz
consuming ""
Exception in thread "main" ExceptionB: ExceptionA: empty
…
Caused by: ExceptionA: empty
…

这里，ExceptionB::new是翻译函数，相当于exA->new ExceptionB(exA)。