是否有Java 8流操作限制流(可能是无限的),直到第一个元素无法匹配谓词?
在Java 9中,我们可以像下面的例子一样使用takeWhile来打印所有小于10的数字。
IntStream
.iterate(1, n -> n + 1)
.takeWhile(n -> n < 10)
.forEach(System.out::println);
因为在Java 8中没有这样的操作,那么以通用的方式实现它的最佳方法是什么呢?
是否有Java 8流操作限制流(可能是无限的),直到第一个元素无法匹配谓词?
在Java 9中,我们可以像下面的例子一样使用takeWhile来打印所有小于10的数字。
IntStream
.iterate(1, n -> n + 1)
.takeWhile(n -> n < 10)
.forEach(System.out::println);
因为在Java 8中没有这样的操作,那么以通用的方式实现它的最佳方法是什么呢?
当前回答
这样的操作在Java 8 Stream中应该是可能的,但它不一定能高效地完成——例如,您不一定能将这样的操作并行化,因为您必须按顺序查看元素。
API并没有提供简单的方法,但最简单的方法可能是获取Stream. Iterator(),将迭代器包装为“take-while”实现,然后返回到Spliterator,然后是Stream。或者——可能——包装Spliterator,尽管在这个实现中它不能再被拆分了。
下面是一个未测试的takeWhile在Spliterator上的实现:
static <T> Spliterator<T> takeWhile(
Spliterator<T> splitr, Predicate<? super T> predicate) {
return new Spliterators.AbstractSpliterator<T>(splitr.estimateSize(), 0) {
boolean stillGoing = true;
@Override public boolean tryAdvance(Consumer<? super T> consumer) {
if (stillGoing) {
boolean hadNext = splitr.tryAdvance(elem -> {
if (predicate.test(elem)) {
consumer.accept(elem);
} else {
stillGoing = false;
}
});
return hadNext && stillGoing;
}
return false;
}
};
}
static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) {
return StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false);
}
其他回答
这样的操作在Java 8 Stream中应该是可能的,但它不一定能高效地完成——例如,您不一定能将这样的操作并行化,因为您必须按顺序查看元素。
API并没有提供简单的方法,但最简单的方法可能是获取Stream. Iterator(),将迭代器包装为“take-while”实现,然后返回到Spliterator,然后是Stream。或者——可能——包装Spliterator,尽管在这个实现中它不能再被拆分了。
下面是一个未测试的takeWhile在Spliterator上的实现:
static <T> Spliterator<T> takeWhile(
Spliterator<T> splitr, Predicate<? super T> predicate) {
return new Spliterators.AbstractSpliterator<T>(splitr.estimateSize(), 0) {
boolean stillGoing = true;
@Override public boolean tryAdvance(Consumer<? super T> consumer) {
if (stillGoing) {
boolean hadNext = splitr.tryAdvance(elem -> {
if (predicate.test(elem)) {
consumer.accept(elem);
} else {
stillGoing = false;
}
});
return hadNext && stillGoing;
}
return false;
}
};
}
static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> predicate) {
return StreamSupport.stream(takeWhile(stream.spliterator(), predicate), false);
}
这是从JDK 9 java.util.stream.Stream.takeWhile(Predicate)中复制的源代码。为了使用JDK 8,有一点不同。
static <T> Stream<T> takeWhile(Stream<T> stream, Predicate<? super T> p) {
class Taking extends Spliterators.AbstractSpliterator<T> implements Consumer<T> {
private static final int CANCEL_CHECK_COUNT = 63;
private final Spliterator<T> s;
private int count;
private T t;
private final AtomicBoolean cancel = new AtomicBoolean();
private boolean takeOrDrop = true;
Taking(Spliterator<T> s) {
super(s.estimateSize(), s.characteristics() & ~(Spliterator.SIZED | Spliterator.SUBSIZED));
this.s = s;
}
@Override
public boolean tryAdvance(Consumer<? super T> action) {
boolean test = true;
if (takeOrDrop && // If can take
(count != 0 || !cancel.get()) && // and if not cancelled
s.tryAdvance(this) && // and if advanced one element
(test = p.test(t))) { // and test on element passes
action.accept(t); // then accept element
return true;
} else {
// Taking is finished
takeOrDrop = false;
// Cancel all further traversal and splitting operations
// only if test of element failed (short-circuited)
if (!test)
cancel.set(true);
return false;
}
}
@Override
public Comparator<? super T> getComparator() {
return s.getComparator();
}
@Override
public void accept(T t) {
count = (count + 1) & CANCEL_CHECK_COUNT;
this.t = t;
}
@Override
public Spliterator<T> trySplit() {
return null;
}
}
return StreamSupport.stream(new Taking(stream.spliterator()), stream.isParallel()).onClose(stream::close);
}
甚至我也有类似的需求——调用web服务,如果失败,重试3次。如果在多次尝试后仍然失败,请发送电子邮件通知。在谷歌搜索了很多之后,anyMatch()成为了救星。我的示例代码如下。在下面的例子中,如果webServiceCall方法在第一次迭代本身中返回true,则stream不会在我们调用anyMatch()时进一步迭代。我相信,这就是你想要的。
import java.util.stream.IntStream;
import io.netty.util.internal.ThreadLocalRandom;
class TrialStreamMatch {
public static void main(String[] args) {
if(!IntStream.range(1,3).anyMatch(integ -> webServiceCall(integ))){
//Code for sending email notifications
}
}
public static boolean webServiceCall(int i){
//For time being, I have written a code for generating boolean randomly
//This whole piece needs to be replaced by actual web-service client code
boolean bool = ThreadLocalRandom.current().nextBoolean();
System.out.println("Iteration index :: "+i+" bool :: "+bool);
//Return success status -- true or false
return bool;
}
allMatch()是一个短路函数,因此可以使用它来停止处理。主要的缺点是您必须进行两次测试:一次是查看是否应该处理它,另一次是查看是否继续进行。
IntStream
.iterate(1, n -> n + 1)
.peek(n->{if (n<10) System.out.println(n);})
.allMatch(n->n < 10);
我有另一个快速的解决方案来实现这个(实际上是不干净的,但你知道的):
public static void main(String[] args) {
System.out.println(StreamUtil.iterate(1, o -> o + 1).terminateOn(15)
.map(o -> o.toString()).collect(Collectors.joining(", ")));
}
static interface TerminatedStream<T> {
Stream<T> terminateOn(T e);
}
static class StreamUtil {
static <T> TerminatedStream<T> iterate(T seed, UnaryOperator<T> op) {
return new TerminatedStream<T>() {
public Stream<T> terminateOn(T e) {
Builder<T> builder = Stream.<T> builder().add(seed);
T current = seed;
while (!current.equals(e)) {
current = op.apply(current);
builder.add(current);
}
return builder.build();
}
};
}
}