Java中的功能接口不声明任何已检查或未检查的异常。 我们需要改变方法的签名:

boolean isActive() throws IOException; 
String getNumber() throwsIOException;

To:

boolean isActive();
String getNumber();

或者用try-catch block处理它:

public Set<String> getActiveAccountNumbers() {
  Stream<Account> s =  accounts.values().stream();
  s = s.filter(a -> 
    try{
      a.isActive();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  Stream<String> ss = s.map(a -> 
    try{
      a.getNumber();
    }catch(IOException e){
      throw new RuntimeException(e);
    }
  );
  return ss.collect(Collectors.toSet());
}

另一种选择是编写自定义包装器或使用像ThrowingFunction这样的库。 使用库，我们只需要将依赖项添加到pom.xml:

<dependency>
    <groupId>pl.touk</groupId>
    <artifactId>throwing-function</artifactId>
    <version>1.3</version>
</dependency>

并使用特定的类，如ThrowingFunction, ThrowingConsumer, ThrowingPredicate, ThrowingRunnable, ThrowingSupplier。

代码的最后是这样的:

public Set<String> getActiveAccountNumbers() {
  return accounts.values().stream()
    .filter(ThrowingPredicate.unchecked(Account::isActive))
    .map(ThrowingFunction.unchecked(Account::getNumber))
    .collect(Collectors.toSet());
}

使用#propagate()方法。来自Sam Beran的Java 8 Blog的非guava实现示例:

public class Throwables {
    public interface ExceptionWrapper<E> {
        E wrap(Exception e);
    }

    public static <T> T propagate(Callable<T> callable) throws RuntimeException {
        return propagate(callable, RuntimeException::new);
    }

    public static <T, E extends Throwable> T propagate(Callable<T> callable, ExceptionWrapper<E> wrapper) throws E {
        try {
            return callable.call();
        } catch (RuntimeException e) {
            throw e;
        } catch (Exception e) {
            throw wrapper.wrap(e);
        }
    }
}

你可以通过包装你的lambda来抛出一个未检查的异常，然后在终端操作中解开这个未检查的异常，从而滚动你自己的Stream变体:

@FunctionalInterface
public interface ThrowingPredicate<T, X extends Throwable> {
    public boolean test(T t) throws X;
}

@FunctionalInterface
public interface ThrowingFunction<T, R, X extends Throwable> {
    public R apply(T t) throws X;
}

@FunctionalInterface
public interface ThrowingSupplier<R, X extends Throwable> {
    public R get() throws X;
}

public interface ThrowingStream<T, X extends Throwable> {
    public ThrowingStream<T, X> filter(
            ThrowingPredicate<? super T, ? extends X> predicate);

    public <R> ThrowingStream<T, R> map(
            ThrowingFunction<? super T, ? extends R, ? extends X> mapper);

    public <A, R> R collect(Collector<? super T, A, R> collector) throws X;

    // etc
}

class StreamAdapter<T, X extends Throwable> implements ThrowingStream<T, X> {
    private static class AdapterException extends RuntimeException {
        public AdapterException(Throwable cause) {
            super(cause);
        }
    }

    private final Stream<T> delegate;
    private final Class<X> x;

    StreamAdapter(Stream<T> delegate, Class<X> x) {
        this.delegate = delegate;
        this.x = x;
    }

    private <R> R maskException(ThrowingSupplier<R, X> method) {
        try {
            return method.get();
        } catch (Throwable t) {
            if (x.isInstance(t)) {
                throw new AdapterException(t);
            } else {
                throw t;
            }
        }
    }

    @Override
    public ThrowingStream<T, X> filter(ThrowingPredicate<T, X> predicate) {
        return new StreamAdapter<>(
                delegate.filter(t -> maskException(() -> predicate.test(t))), x);
    }

    @Override
    public <R> ThrowingStream<R, X> map(ThrowingFunction<T, R, X> mapper) {
        return new StreamAdapter<>(
                delegate.map(t -> maskException(() -> mapper.apply(t))), x);
    }

    private <R> R unmaskException(Supplier<R> method) throws X {
        try {
            return method.get();
        } catch (AdapterException e) {
            throw x.cast(e.getCause());
        }
    }

    @Override
    public <A, R> R collect(Collector<T, A, R> collector) throws X {
        return unmaskException(() -> delegate.collect(collector));
    }
}

然后你可以像使用流一样使用它:

Stream<Account> s = accounts.values().stream();
ThrowingStream<Account, IOException> ts = new StreamAdapter<>(s, IOException.class);
return ts.filter(Account::isActive).map(Account::getNumber).collect(toSet());

这个解决方案需要相当多的样板文件，所以我建议您看一看我已经创建的库，它完全符合我在这里为整个Stream类(以及更多!)所描述的内容。

为了正确地添加IOException(到RuntimeException)处理代码，你的方法看起来像这样:

Stream<Account> s =  accounts.values().stream();

s = s.filter(a -> { try { return a.isActive(); } 
  catch (IOException e) { throw new RuntimeException(e); }});

Stream<String> ss = s.map(a -> { try { return a.getNumber() }
  catch (IOException e) { throw new RuntimeException(e); }});

return ss.collect(Collectors.toSet());

现在的问题是，IOException必须被捕获为RuntimeException并转换回IOException——这将向上述方法添加更多代码。

为什么要使用Stream，因为它可以这样做——而且该方法会抛出IOException，所以也不需要额外的代码:

Set<String> set = new HashSet<>();
for(Account a: accounts.values()){
  if(a.isActive()){
     set.add(a.getNumber());
  } 
}
return set;

也可以使用一些外部(流)错误指示器在更高级别抛出异常:

List<String> errorMessages = new ArrayList<>(); // error indicator
//..
errorMessages.clear();

List<String> names = new ArrayList<>(Arrays.asList("andrey", "angela", "pamela"));

names.stream()
.map(name -> {
    if (name != "pamela") {
      errorMessages.add(name + " is wrong here!"); 
      return null; // triggering the indicator
    }
    return name;
} )
.filter(elem -> (elem != null)) // bypassing propagation of only current unwanted data
//.filter(elem -> (errorMessages.size() == 0)) // or blocking any propagation once unwanted data detected
.forEach(System.out::println);

if (errorMessages.size() > 0) { // handling the indicator
  throw  new RuntimeException(String,join(", ", errorMessages));
}

如果你不介意使用第三方库，AOL的cyclops-react库，disclosure::我是一个贡献者，有一个ExceptionSoftener类可以在这里提供帮助。

 s.filter(softenPredicate(a->a.isActive()));