Java 8 streams条件处理
我感兴趣的是将一个流分成两个或多个子流，并以不同的方式处理元素。例如，一个(大)文本文件可能包含类型为A的行和类型为B的行，在这种情况下，我希望这样做：
File.lines(path)
.filter(line -> isTypeA(line))
.forEachTrue(line -> processTypeA(line))
.forEachFalse(line -> processTypeB(line))前一个是我对情况进行抽象的尝试。实际上，我有一个非常大的文本文件，其中的每一行都在针对一个正则表达式进行测试；如果该行通过，那么它就会被处理，而如果它被拒绝，那么我想要更新一个计数器。这种对被拒绝字符串的进一步处理是我不简单地使用filter的原因。
有没有什么合理的方法可以用streams来做到这一点，或者我必须回退到循环？(我希望这也能并行运行，所以streams是我的首选)。
回答 5
Stack Overflow用户
发布于 2016-03-08 18:54:35
虽然行为参数中的副作用是不鼓励的，但只要没有干扰，它们就不会被禁止，所以最简单但不是最干净的解决方案是在过滤器中进行计数：
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，它可以基于一个条件组合两个collect操作的处理。这要求您能够将下游操作表示为收集器，但大多数流操作可以表示为收集器(并且趋势是以这种方式表示所有操作的可能性，即Java9将添加当前缺少的filtering和flatMapping。
您需要一个pair类型来保存两个结果，因此假设草图如下所示
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;收集器是并行友好的，允许任意复杂的委托收集器，但请注意，与“Reader#lines() parallelizes badly due to nonconfigurable batch size policy in its spliterator”相比，在当前实现中，Files.lines返回的流在并行处理时可能表现不佳。计划在Java版本9中进行改进。
Stack Overflow用户
发布于 2016-03-08 05:47:28
我处理这个问题的方式根本不是拆分，而是写
Files.lines(path)
.map(line -> {
if (condition(line)) {
return doThingA(line);
} else {
return doThingB(line);
}
})...具体细节取决于你到底想做什么，以及你打算怎么做。
Stack Overflow用户
发布于 2018-07-26 08:58:47
这里有一种方法(它忽略了强制条件处理到流中的注意事项)，它将谓词和消费者包装到一个有副作用的单个谓词中：
public static class StreamProc {
public static <T> Predicate<T> process( Predicate<T> condition, Consumer<T> operation ) {
Predicate<T> p = t -> { operation.accept(t); return false; };
return (t) -> condition.test(t) ? p.test(t) : true;
}
}然后对流进行过滤：
someStream
.filter( StreamProc.process( cond1, op1 ) )
.filter( StreamProc.process( cond2, op2 ) )
...
.collect( ... )流中剩余的元素尚未处理。
例如，使用外部迭代的典型文件系统遍历如下所示
File[] files = dir.listFiles();
for ( File f : files ) {
if ( f.isDirectory() ) {
this.processDir( f );
} else if ( f.isFile() ) {
this.processFile( f );
} else {
this.processErr( f );
}
}有了流和内部迭代，这就变成了
Arrays.stream( dir.listFiles() )
.filter( StreamProc.process( f -> f.isDirectory(), this::processDir ) )
.filter( StreamProc.process( f -> f.isFile(), this::processFile ) )
.forEach( f -> this::processErr );我希望Stream直接实现process方法。然后我们就可以
Arrays.stream( dir.listFiles() )
.process( f -> f.isDirectory(), this::processDir ) )
.process( f -> f.isFile(), this::processFile ) )
.forEach( f -> this::processErr );有什么想法？
https://stackoverflow.com/questions/35854574
复制相似问题
腾讯云开发者
