procesamiento - stream java collect

¿Por qué se opera este Java Stream dos veces?

Question

La API de Java 8 dice:

El recorrido de la fuente de la tubería no comienza hasta que se ejecuta la operación terminal de la tubería.

Entonces, ¿por qué arroja el siguiente código:

java.lang.IllegalStateException: la transmisión ya ha sido operada o cerrada

Stream<Integer> stream = Stream.of(1,2,3);
stream.filter( x-> x>1 );
stream.filter( x-> x>2 ).forEach(System.out::print);

No se supone que la primera operación de filtrado según la API opere en Stream.

Answer 1

Este es un mal uso de la stream que se detecta cuando le adjunta más de un .fliter() .

No dice que ha sido atravesado más de una vez, solo que "ya ha sido operado"

Answer 2

Esto sucede porque ignora el valor de retorno del filter .

Stream<Integer> stream = Stream.of(1,2,3);
stream.filter( x-> x>1 ); // <-- ignoring the return value here
stream.filter( x-> x>2 ).forEach(System.out::print);

Stream.filter devuelve una nueva Stream consta de los elementos de esta secuencia que coinciden con el predicado dado. Pero es importante tener en cuenta que es un nuevo Stream. El viejo ha sido operado cuando se le agregó el filtro. Pero el nuevo no lo era.

Citando de Stream Javadoc:

Se debe operar una secuencia (invocando una operación de secuencia intermedia o terminal) solo una vez.

En este caso, el filter es la operación intermedia que operaba en la antigua instancia de Stream.

Entonces este código funcionará bien:

Stream<Integer> stream = Stream.of(1,2,3);
stream = stream.filter( x-> x>1 ); // <-- NOT ignoring the return value here
stream.filter( x-> x>2 ).forEach(System.out::print);

Como señaló Brian Goetz, comúnmente encadenaría esas llamadas:

Stream.of(1,2,3).filter( x-> x>1 )
                .filter( x-> x>2 )
                .forEach(System.out::print);

Answer 3

La documentación en Streams dice:

"Una transmisión debe operarse (invocando una operación de transmisión intermedia o terminal) solo una vez".

De hecho, puedes ver esto en el código fuente. Cuando llama al filtro, devuelve una nueva operación sin estado, pasando la instancia de canalización actual en el constructor ( this ):

@Override
public final Stream<P_OUT> filter(Predicate<? super P_OUT> predicate) {
    Objects.requireNonNull(predicate);
    return new StatelessOp<P_OUT, P_OUT>(this, StreamShape.REFERENCE,
                                       StreamOpFlag.NOT_SIZED) {
        ....
}

La llamada al constructor termina llamando al constructor AbstractPipeline , que se configura así:

AbstractPipeline(AbstractPipeline<?, E_IN, ?> previousStage, int opFlags) {
    if (previousStage.linkedOrConsumed)
        throw new IllegalStateException(MSG_STREAM_LINKED);
    previousStage.linkedOrConsumed = true;
    ...
}

La primera vez que llama al filtro en la fuente (línea 2), establece el valor booleano en verdadero. Como no reutiliza el valor de retorno dado por filtro, la segunda llamada al filtro (línea 3) detectará que la fuente de flujo original (línea 1) ya se ha vinculado (debido a la primera llamada de filtro) y, por lo tanto, la excepción es obtener.

Answer 4

filter() método filter() utiliza la secuencia y devuelve otra instancia de Stream que ignora en su ejemplo.

filter es una operación intermedia pero no puede llamar a filter dos veces en la misma instancia de flujo

Su código debe escribirse de la siguiente manera:

Stream<Integer> stream = Stream.of(1,2,3);
                               .filter( x-> x>1 )
                               .filter( x-> x>2);
stream.forEach(System.out::print);

Como el filtro es una operación intermedia, no se hace "nada" al llamar a estos métodos. Todo el trabajo se procesa realmente cuando se llama al método forEach()