Java 8流条件处理

Question

我有兴趣将一个流分成两个或多个子流,并以不同的方式处理这些元素.例如,(大)文本文件可能包含类型A的行和类型B的行,在这种情况下,我想要执行以下操作:

File.lines(path)
.filter(line -> isTypeA(line))
.forEachTrue(line -> processTypeA(line))
.forEachFalse(line -> processTypeB(line))

以前是我试图抽象的情况.实际上我有一个非常大的文本文件,其中每一行都在测试正则表达式; 如果该行通过,则处理它,而如果它被拒绝,那么我想更新一个计数器.对拒绝字符串的进一步处理是我不仅仅使用的原因filter.

是否有任何合理的方法可以使用流来执行此操作,还是必须回退到循环？(我希望这也是并行运行,所以溪流是我的第一选择).

Answer 1

Java 8流不是为支持这种操作而设计的.来自jdk:

应该仅对一个流进行操作(调用中间或终端流操作).例如,这排除了"分叉"流,其中相同的源提供两个或更多个管道,或者同一个流的多个遍历.

如果你可以将它存储在内存中,你可以使用它,Collectors.partitioningBy如果你只有两种类型并使用Map<Boolean, List>.否则使用Collectors.groupingBy.

Answer 2

只需测试每个元素,并采取相应的行动.

lines.forEach(line -> {
    if (isTypeA(line)) processTypeA(line);
    else processTypeB(line);
});

此行为可能隐藏在辅助方法中:

public static <T> Consumer<T> branch(Predicate<? super T> test, 
                                     Consumer<? super T> t, 
                                     Consumer<? super T> f) {
    return o -> {
        if (test.test(o)) t.accept(o);
        else f.accept(o);
    };
}

然后用法如下:

lines.forEach(branch(this::isTypeA, this::processTypeA, this::processTypeB));

切线说明

该Files.lines()方法不会关闭基础文件,因此您必须使用它:

try (Stream<String> lines = Files.lines(path, encoding)) {
  lines.forEach(...);
}

Stream类型的变量为我抛出一点红旗,所以我更喜欢BufferedReader直接管理:

try (BufferedReader lines = Files.newBufferedReader(path, encoding)) {
    lines.lines().forEach(...);
}

Answer 3

虽然不鼓励使用行为参数中的副作用,但只要不存在干扰,就不会禁止它们,因此最简单但不是最干净的解决方案是在过滤器中计算:

AtomicInteger rejected=new AtomicInteger();
Files.lines(path)
    .filter(line -> {
        boolean accepted=isTypeA(line);
        if(!accepted) rejected.incrementAndGet();
        return accepted;
})
// chain processing of matched lines

只要您处理所有项目,结果将是一致的.只有在使用短路终端操作(并行流)时,结果才会变得不可预测.

更新原子变量可能不是最有效的解决方案,但在处理来自文件的行的上下文中,开销可能可以忽略不计.

如果您想要一个干净,并行友好的解决方案,一种通用的方法是实现一个Collector可以根据条件组合两个收集操作的处理.这要求您能够将下游操作表示为收集器,但大多数流操作可以表示为收集器(趋势是可能以这种方式表达所有操作,即Java 9将添加当前缺失的filtering和flatMapping.

你需要一个对类型来保存两个结果,所以假设一个草图

class Pair<A,B> {
    final A a;
    final B b;
    Pair(A a, B b) {
        this.a=a;
        this.b=b;
    }
}

组合收集器实现看起来像

public static <T, A1, A2, R1, R2> Collector<T, ?, Pair<R1,R2>> conditional(
        Predicate<? super T> predicate,
        Collector<T, A1, R1> whenTrue, Collector<T, A2, R2> whenFalse) {
    Supplier<A1> s1=whenTrue.supplier();
    Supplier<A2> s2=whenFalse.supplier();
    BiConsumer<A1, T> a1=whenTrue.accumulator();
    BiConsumer<A2, T> a2=whenFalse.accumulator();
    BinaryOperator<A1> c1=whenTrue.combiner();
    BinaryOperator<A2> c2=whenFalse.combiner();
    Function<A1,R1> f1=whenTrue.finisher();
    Function<A2,R2> f2=whenFalse.finisher();
    return Collector.of(
        ()->new Pair<>(s1.get(), s2.get()),
        (p,t)->{
            if(predicate.test(t)) a1.accept(p.a, t); else a2.accept(p.b, t);
        },
        (p1,p2)->new Pair<>(c1.apply(p1.a, p2.a), c2.apply(p1.b, p2.b)),
        p -> new Pair<>(f1.apply(p.a), f2.apply(p.b)));
}

并且可以用于例如将匹配项目收集到列表中并计算不匹配项,如下所示:

Pair<List<String>, Long> p = Files.lines(path)
  .collect(conditional(line -> isTypeA(line), Collectors.toList(), Collectors.counting()));
List<String> matching=p.a;
long nonMatching=p.b;

收集器是并行友好的,并允许任意复杂的委托收集器,但请注意,对于当前实现,返回的流Files.lines可能在并行处理方面表现不佳,与"Reader#lines()并行化由于不可配置的批处理大小策略而导致严重并行化它的分裂者".Java 9发行版计划进行了改进.