流式传输时重新引发异常

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)

createRethrowingIterator 包装原始迭代器并返回实际上将 ExceptionA 重新抛出到 ExceptionB 的新迭代器

不是我想要的，因为stream.iterator() 是终端操作符，即它会消耗流，如果流真的很大，可能会导致内存问题

Answer 1

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

它还通过将方法characteristics() 和estimateSize() 委托给源来打开性能改进的机会，因为异常转换功能不会改变它们。您还可以获得不错的并行支持，通过委托源实现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)。

Answer 2

你为什么不包装那个将你的ExceptionA 抛出的调用到一个映射函数中，如果抛出它就会立即将它转换为ExceptionB，比如：

try {
  List<T> l = stream.map(o -> wrapped(() -> o.whateverThrowsExceptionA())).collect(toList());
  // or do your stream2 operations first, before collecting the list
} catch (ExceptionB b) {
  // handle your exception
}

wrapped 在这种情况下类似于：

<T> T wrapped(Callable<T> o) throws ExceptionB {
  try {
    return callable.call();
  } catch (Exception e) {
    throw new ExceptionB(e);
  }
}

您甚至可能想要调整包装器以接收自定义的ExceptionA-catch 函数。

Answer 3

好吧，我没看懂终端操作不代表流完全消耗。感谢 Louis Wasserman 澄清这一点。

为了演示它，我编写了一些单元测试：

import org.junit.Test;

import java.util.Iterator;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

/**
 * @author Beka Tsotsoria
 */
public class StreamExceptionRethrowingTest {

    @Test
    public void throwingIteratorMustBeConsumedWhenStreamIsCollected() throws Exception {
        ThrowingIterator itToBeConsumed = new ThrowingIterator();

        assertThatThrownBy(() -> streamFromIterator(itToBeConsumed)
            .collect(Collectors.toList()))
            .isInstanceOf(ExceptionA.class);

        assertThat(itToBeConsumed.consumed()).isTrue();
    }

    @Test
    public void throwingIteratorMustNotBeConsumedUntilNewStreamIsCollected() throws Exception {
        ThrowingIterator itNotToBeConsumed = new ThrowingIterator();
        RethrowingIterator rethrowingIterator = new RethrowingIterator(streamFromIterator(itNotToBeConsumed).iterator());

        assertThat(itNotToBeConsumed.consumed()).isFalse();

        Stream<String> stream2 = streamFromIterator(rethrowingIterator);

        assertThat(itNotToBeConsumed.consumed()).isFalse();

        assertThatThrownBy(() -> stream2
            .collect(Collectors.toList()))
            .hasCauseInstanceOf(ExceptionA.class)
            .isInstanceOf(ExceptionB.class);

        assertThat(itNotToBeConsumed.consumed()).isTrue();
    }

    @Test
    public void streamIteratorMustNotBeConsumedUntilNewStreamIsCollected() throws Exception {
        Stream<String> stream = Stream.of("dummy")
            .map(d -> {
                throw new ExceptionA();
            });

        Stream<String> stream2 = streamFromIterator(new RethrowingIterator(stream.iterator()));
        // No exceptions so far, i.e. stream.iterator() was not consumed

        assertThatThrownBy(() -> stream2
            .collect(Collectors.toList()))
            .hasCauseInstanceOf(ExceptionA.class)
            .isInstanceOf(ExceptionB.class);
    }

    private Stream<String> streamFromIterator(Iterator<String> it) {
        return StreamSupport.stream(Spliterators.spliteratorUnknownSize(it, Spliterator.NONNULL), false);
    }

    static class ThrowingIterator implements Iterator<String> {

        private boolean hasNextCalled;
        private boolean nextCalled;

        @Override
        public boolean hasNext() {
            hasNextCalled = true;
            throw new ExceptionA();
        }

        @Override
        public String next() {
            nextCalled = true;
            throw new ExceptionA();
        }

        public boolean consumed() {
            return hasNextCalled || nextCalled;
        }
    }

    static class RethrowingIterator implements Iterator<String> {

        private Iterator<String> it;

        public RethrowingIterator(Iterator<String> it) {
            this.it = it;
        }

        @Override
        public boolean hasNext() {
            try {
                return it.hasNext();
            } catch (ExceptionA e) {
                throw new ExceptionB(e);
            }
        }

        @Override
        public String next() {
            try {
                return it.next();
            } catch (ExceptionA e) {
                throw new ExceptionB(e);
            }
        }
    }


    static class ExceptionA extends RuntimeException {

    }

    static class ExceptionB extends RuntimeException {

        public ExceptionB(Throwable cause) {
            super(cause);
        }
    }
}    

感谢您的 cmets。干杯！