java 자바 - 스트림을 복사하여 "스트림이 이미 작동 중이거나 닫혀 있음"을 방지하십시오.





stream println (8)


java.util.function.Supplier 사용하십시오.

http://winterbe.com/posts/2014/07/31/java8-stream-tutorial-examples/ :

스트림 재사용

Java 8 스트림은 재사용 할 수 없습니다. 터미널 작업을 호출하자마자 스트림이 닫힙니다.

Stream<String> stream =

Stream.of("d2", "a2", "b1", "b3", "c")

.filter(s -> s.startsWith("a"));

stream.anyMatch(s -> true);    // ok

stream.noneMatch(s -> true);   // exception

같은 스트림상의 anyMatch의 후에 noneMatch를 호출하면 (자), 다음의 예외가 throw됩니다.

java.lang.IllegalStateException: stream has already been operated upon or closed

at 

java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:229)

at 

java.util.stream.ReferencePipeline.noneMatch(ReferencePipeline.java:459)

at com.winterbe.java8.Streams5.test7(Streams5.java:38)

at com.winterbe.java8.Streams5.main(Streams5.java:28)

이 제한을 극복하기 위해, 우리는 실행하고자하는 모든 터미널 작업에 대해 새로운 스트림 체인을 생성해야합니다. 예를 들어 모든 중간 작업이 이미 설정된 상태에서 새로운 스트림을 생성하기 위해 스트림 공급 업체를 생성 할 수 있습니다.

Supplier<Stream<String>> streamSupplier =

    () -> Stream.of("d2", "a2", "b1", "b3", "c")

            .filter(s -> s.startsWith("a"));

streamSupplier.get().anyMatch(s -> true);   // ok

streamSupplier.get().noneMatch(s -> true);  // ok

get() 호출 할 get() 마다 원하는 터미널 작업을 호출하기 위해 저장 한 새 스트림이 생성됩니다.

Java 2 스트림을 복제하여 두 번 처리 할 수 ​​있습니다. 목록으로 collect 하고 그 목록에서 새 스트림을 얻을 수 있습니다.

// doSomething() returns a stream
List<A> thing = doSomething().collect(toList());
thing.stream()... // do stuff
thing.stream()... // do other stuff

그러나 좀 더 효율적이고 우아한 방법이 있어야한다고 생각합니다.

스트림을 콜렉션으로 변환하지 않고 스트림을 복사하는 방법이 있습니까?

저는 실제로 Either 의 스트림으로 작업하기 때문에 왼쪽 프로젝션을 한 방향으로 처리하기 전에 오른쪽 프로젝션으로 이동하고 다른 방법으로 처리하고 싶습니다. 이런 종류의 (이것은 지금까지 toList 트릭을 사용해야 만했습니다.)

List<Either<Pair<A, Throwable>, A>> results = doSomething().collect(toList());

Stream<Pair<A, Throwable>> failures = results.stream().flatMap(either -> either.left());
failures.forEach(failure -> ... );

Stream<A> successes = results.stream().flatMap(either -> either.right());
successes.forEach(success -> ... );



우리는 jOOλ 에 대한 통합 테스트를 개선하기 위해 만든 오픈 소스 라이브러리 인 jOOλ 스트림에 대해 duplicate() 메소드를 구현했습니다. 본질적으로 다음과 같이 작성할 수 있습니다.

Tuple2<Seq<A>, Seq<A>> duplicates = Seq.seq(doSomething()).duplicate();

내부적으로 한 스트림에서 소비되었지만 다른 스트림에서 소비 된 모든 값을 저장하는 버퍼가 있습니다. 두 스트림을 같은 속도로 소비하고 스레드 안전성이 부족한 상태에서 살 수 있다면 효과적 일 것입니다.

다음은 알고리즘 작동 방식입니다.

static <T> Tuple2<Seq<T>, Seq<T>> duplicate(Stream<T> stream) {
    final List<T> gap = new LinkedList<>();
    final Iterator<T> it = stream.iterator();

    @SuppressWarnings("unchecked")
    final Iterator<T>[] ahead = new Iterator[] { null };

    class Duplicate implements Iterator<T> {
        @Override
        public boolean hasNext() {
            if (ahead[0] == null || ahead[0] == this)
                return it.hasNext();

            return !gap.isEmpty();
        }

        @Override
        public T next() {
            if (ahead[0] == null)
                ahead[0] = this;

            if (ahead[0] == this) {
                T value = it.next();
                gap.offer(value);
                return value;
            }

            return gap.poll();
        }
    }

    return tuple(seq(new Duplicate()), seq(new Duplicate()));
}

더 많은 소스 코드보기

Tuple2Pair 유형과 비슷하지만 Seq 은 일부 기능이 향상된 Stream 입니다.




runnables의 스트림을 생성 할 수 있습니다 (예 :).

results.stream()
    .flatMap(either -> Stream.<Runnable> of(
            () -> failure(either.left()),
            () -> success(either.right())))
    .forEach(Runnable::run);

failuresuccess 이 적용 할 작업입니다. 그러나 이것은 꽤 많은 임시 객체를 생성 할 것이고 콜렉션에서 시작하고 두 번 스트리밍 / 반복하는 것보다 효율적이지 않을 수도 있습니다.




공급 업체를 사용하여 각 종단 작업에 대한 스트림을 생산하십시오.

Supplier <Stream<Integer>> streamSupplier=()->list.stream();

컬렉션의 스트림이 필요할 때마다 streamSupplier.get() 을 사용하여 새 스트림을 가져 오십시오.

예 :

  1. streamSupplier.get().anyMatch(predicate);
  2. streamSupplier.get().allMatch(predicate2);



이 특별한 문제에 대해서도 파티셔닝을 사용할 수 있습니다. 좋아하는 것

     // Partition Eighters into left and right
     List<Either<Pair<A, Throwable>, A>> results = doSomething();
     Map<Boolean, Object> passingFailing = results.collect(Collectors.partitioningBy(s -> s.isLeft()));
     passingFailing.get(true) <- here will be all passing (left values)
     passingFailing.get(false) <- here will be all failing (right values)



효율성에 대한 귀하의 가정은 거꾸로 생각됩니다. 데이터를 저장할 필요가 없기 때문에 데이터를 한 번만 사용하면 효율성이 크게 향상되며 스트림은 파이프 라인을 통해 전체 데이터를 효율적으로 전달할 수있는 강력한 "루프 융합"최적화를 제공합니다.

동일한 데이터를 다시 사용하려면 정의에 따라 두 번 (결정 론적으로) 생성하거나 저장해야합니다. 이미 컬렉션에있는 경우, 훌륭합니다. 두 번 반복하면 저렴합니다.

우리는 "포크 스트림"으로 디자인을 실험했습니다. 우리가 발견 한 것은 이것을 지원하는 것이 실질적인 비용을 들이고 있다는 것입니다. 일반적이지 않은 경우를 희생시키면서 일반 사례 (한 번 사용)에 부담을 줬다. 큰 문제는 "두 파이프 라인이 동일한 속도로 데이터를 소비하지 않을 때 어떤 일이 발생하는지"를 다루는 것이 었습니다. 이제 어쨌든 버퍼링으로 돌아 왔습니다. 이것은 분명히 무게를 지니지 않는 특징이었습니다.

동일한 데이터를 반복해서 사용하려면 해당 데이터를 저장하거나 고객을 조작자로 구성하고 다음을 수행하십시오.

stream()...stuff....forEach(e -> { consumerA(e); consumerB(e); });

RxJava 라이브러리의 프로세싱 모델은 이러한 종류의 "스트림 포킹 (stream forking)"에 더 도움이되기 때문에 RxJava 라이브러리를 살펴볼 수도 있습니다.




내가 기여한 라이브러리 인 cyclops-react 는 스트림을 복제하고 스트림의 jOOλ 튜플을 반환하는 정적 메서드를 가지고 있습니다.

    Stream<Integer> stream = Stream.of(1,2,3);
    Tuple2<Stream<Integer>,Stream<Integer>> streams =  StreamUtils.duplicate(stream);

기존 스트림에서 복제본을 사용할 때 발생하는 성능상의 불이익이 있습니다. 보다 효과적인 대안은 Streamable을 사용하는 것입니다.

Stream, Iterable 또는 Array로 구성 할 수 있고 여러 번 재생할 수있는 (게으른) Streamable 클래스가 있습니다.

    Streamable<Integer> streamable = Streamable.of(1,2,3);
    streamable.stream().forEach(System.out::println);
    streamable.stream().forEach(System.out::println);

AsStreamable.synchronizedFromStream (stream) - 스레드간에 공유 될 수있는 방식으로 지연 컬렉션을 지연 생성하는 Streamable을 만드는 데 사용할 수 있습니다. Streamable.fromStream (stream)은 동기화 오버 헤드를 발생시키지 않습니다.




T[] 는 일부 일반 객체의 배열을 나타냅니다. 모든 배열 유형 ( int[] )은 객체입니다. 따라서 int[][]T = int[] 때 유효한 T[] T = int[] 입니다.

그러나 int 는 객체가 아니기 때문에 int int[][] 는 유효한 T[][] 가 아닙니다.





java lambda java-8 java-stream