java - studio - La manera más eficiente de obtener el último elemento de una transmisión

new icon java

Question

Stream no tiene un last() método last() :

Stream<T> stream;
T last = stream.last(); // No such method

¿Cuál es la forma más elegante y / o eficiente de obtener el último elemento (o nulo para un flujo vacío)?

Answer 1

Aquí hay otra solución (no tan eficiente):

List<String> list = Arrays.asList("abc","ab","cc");
long count = list.stream().count();
list.stream().skip(count-1).findFirst().ifPresent(System.out::println);

Answer 2

Creo que esta solución es más eficiente y legible que la solución de Holger :

import java.util.Spliterator;
import static java.util.Spliterator.ORDERED;
import java.util.stream.Stream;

/**
 * @param <T>    the type of elements in the stream
 * @param stream a stream
 * @return the last element in the stream
 * @throws AssertionError if the stream is unordered
 */
public static <T> Optional<T> getLastElement(Stream<T> stream)
{
    Spliterator<T> spliterator = stream.spliterator();
    assert (spliterator.hasCharacteristics(ORDERED)): "Operation makes no sense on unordered streams";

    // First we skip as many elements as possible
    Consumer<T> noOp = input -> {};
    while (true) {
        // trySplit() moves the first spliterator forward by the size of the second spliterator
        Spliterator<T> second = spliterator.trySplit();
        if (second == null)
            break;
        if (!spliterator.tryAdvance(noOp)) {
            // If the first spliterator is empty, continue splitting the second spliterator
            spliterator = second;
        }
    }

    // Then we consume the last element(s)
    LastElementConsumer<T> consumer = new LastElementConsumer<>();
    spliterator.forEachRemaining(consumer);
    return consumer.get();
}

[...]

import java.util.Optional;
import java.util.function.Consumer;

/**
 * A consumer that returns the last value that was consumed.
 * <p>
 * @param <T> the type of elements to consume
 * @author Gili Tzabari
 */
public final class LastElementConsumer<T> implements Consumer<T>
{
    private Optional<T> result = Optional.empty();

    @Override
    public void accept(T t)
    {
        result = Optional.of(t);
    }

    /**
     * @return the last value that was consumed
     */
    public Optional<T> get()
    {
        return result;
    }
}

Si tu corres:

String s = getLastElement(IntStream.range(0, 10_000_000).mapToObj(i->
  {
    System.out.println("Potential heavy operation on " + i);
    return String.valueOf(i);
  }).parallel()
);
System.out.println(s);

imprimirá el mismo resultado que la solución de Holger:

Potential heavy operation on 9999999
9999999

En otras palabras, no realizó la operación en los primeros elementos 9999999, sino solo en la última.

Answer 3

Esto depende en gran medida de la naturaleza de la Stream . Tenga en cuenta que "simple" no necesariamente significa "eficiente". Si sospecha que la transmisión es muy grande, lleva a cabo operaciones pesadas o tiene una fuente que conoce el tamaño por adelantado, lo siguiente puede ser sustancialmente más eficiente que la solución simple:

static <T> T getLast(Stream<T> stream) {
    Spliterator<T> sp=stream.spliterator();
    if(sp.hasCharacteristics(Spliterator.SIZED|Spliterator.SUBSIZED)) {
        for(;;) {
            Spliterator<T> part=sp.trySplit();
            if(part==null) break;
            if(sp.getExactSizeIfKnown()==0) {
                sp=part;
                break;
            }
        }
    }
    T value=null;
    for(Iterator<T> it=recursive(sp); it.hasNext(); )
        value=it.next();
    return value;
}

private static <T> Iterator<T> recursive(Spliterator<T> sp) {
    Spliterator<T> prev=sp.trySplit();
    if(prev==null) return Spliterators.iterator(sp);
    Iterator<T> it=recursive(sp);
    if(it!=null && it.hasNext()) return it;
    return recursive(prev);
}

Puede ilustrar la diferencia con el siguiente ejemplo:

String s=getLast(
    IntStream.range(0, 10_000_000).mapToObj(i-> {
        System.out.println("potential heavy operation on "+i);
        return String.valueOf(i);
    }).parallel()
);
System.out.println(s);

Se imprimirá:

potential heavy operation on 9999999
9999999

En otras palabras, no realizó la operación en los primeros elementos 9999999, sino solo en la última.

Answer 4

Esto es solo una refactorización de la respuesta de porque el código, aunque es fantástico, es un poco difícil de leer / entender, especialmente para las personas que no eran programadores de C antes de Java. Espero que mi clase de ejemplo refactorizado sea un poco más fácil de seguir para aquellos que no están familiarizados con los spliterators, qué hacen o cómo funcionan.

public class LastElementFinderExample {
    public static void main(String[] args){
        String s = getLast(
            LongStream.range(0, 10_000_000_000L).mapToObj(i-> {
                System.out.println("potential heavy operation on "+i);
                return String.valueOf(i);
            }).parallel()
        );
        System.out.println(s);
    }

    public static <T> T getLast(Stream<T> stream){
        Spliterator<T> sp = stream.spliterator();
        if(isSized(sp)) {
            sp = getLastSplit(sp);
        }
        return getIteratorLastValue(getLastIterator(sp));
    }

    private static boolean isSized(Spliterator<?> sp){
        return sp.hasCharacteristics(Spliterator.SIZED|Spliterator.SUBSIZED);
    }

    private static <T> Spliterator<T> getLastSplit(Spliterator<T> sp){
        return splitUntil(sp, s->s.getExactSizeIfKnown() == 0);
    }

    private static <T> Iterator<T> getLastIterator(Spliterator<T> sp) {
        return Spliterators.iterator(splitUntil(sp, null));
    }

    private static <T> T getIteratorLastValue(Iterator<T> it){
        T result = null;
        while (it.hasNext()){
            result = it.next();
        }
        return result;
    }

    private static <T> Spliterator<T> splitUntil(Spliterator<T> sp, Predicate<Spliterator<T>> condition){
        Spliterator<T> result = sp;
        for (Spliterator<T> part = sp.trySplit(); part != null; part = result.trySplit()){
            if (condition == null || condition.test(result)){
                result = part;
            }
        }
        return result;      
    }   
}

Answer 5

Guava tiene Streams.findLast :

Stream<T> stream;
T last = Streams.findLast(stream);

Answer 6

Haga una reducción que simplemente devuelve el valor actual:

Stream<T> stream;
T last = stream.reduce((a, b) -> b).orElse(null);

Answer 7

Las transmisiones paralelas sin clasificar con métodos de "omisión" son complicadas y la implementación de @ Holger da una respuesta incorrecta. También la implementación de @ Holger es un poco más lenta porque usa iteradores.

Una optimización de la respuesta de @Holger:

public static <T> Optional<T> last(Stream<? extends T> stream) {
    Objects.requireNonNull(stream, "stream");

    Spliterator<? extends T> spliterator = stream.spliterator();
    Spliterator<? extends T> lastSpliterator = spliterator;

    // Note that this method does not work very well with:
    // unsized parallel streams when used with skip methods.
    // on that cases it will answer Optional.empty.

    // Find the last spliterator with estimate size
    // Meaningfull only on unsized parallel streams
    if(spliterator.estimateSize() == Long.MAX_VALUE) {
        for (Spliterator<? extends T> prev = spliterator.trySplit(); prev != null; prev = spliterator.trySplit()) {
            lastSpliterator = prev;
        }
    }

    // Find the last spliterator on sized streams
    // Meaningfull only on parallel streams (note that unsized was transformed in sized)
    for (Spliterator<? extends T> prev = lastSpliterator.trySplit(); prev != null; prev = lastSpliterator.trySplit()) {
        if (lastSpliterator.estimateSize() == 0) {
            lastSpliterator = prev;
            break;
        }
    }

    // Find the last element of the last spliterator
    // Parallel streams only performs operation on one element
    AtomicReference<T> last = new AtomicReference<>();
    lastSpliterator.forEachRemaining(last::set);

    return Optional.ofNullable(last.get());
}

Pruebas unitarias usando junit 5:

@Test
@DisplayName("last sequential sized")
void last_sequential_sized() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed();
    stream = stream.skip(50_000).peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(9_950_000L);
}

@Test
@DisplayName("last sequential unsized")
void last_sequential_unsized() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed();
    stream = StreamSupport.stream(((Iterable<Long>) stream::iterator).spliterator(), stream.isParallel());
    stream = stream.skip(50_000).peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(9_950_000L);
}

@Test
@DisplayName("last parallel sized")
void last_parallel_sized() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed().parallel();
    stream = stream.skip(50_000).peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(1);
}

@Test
@DisplayName("getLast parallel unsized")
void last_parallel_unsized() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed().parallel();
    stream = StreamSupport.stream(((Iterable<Long>) stream::iterator).spliterator(), stream.isParallel());
    stream = stream.peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(1);
}

@Test
@DisplayName("last parallel unsized with skip")
void last_parallel_unsized_with_skip() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed().parallel();
    stream = StreamSupport.stream(((Iterable<Long>) stream::iterator).spliterator(), stream.isParallel());
    stream = stream.skip(50_000).peek(num -> count.getAndIncrement());

    // Unfortunately unsized parallel streams does not work very well with skip
    //assertThat(Streams.last(stream)).hasValue(expected);
    //assertThat(count).hasValue(1);

    // @Holger implementation gives wrong answer!!
    //assertThat(Streams.getLast(stream)).hasValue(9_950_000L); //!!!
    //assertThat(count).hasValue(1);

    // This is also not a very good answer better
    assertThat(Streams.last(stream)).isEmpty();
    assertThat(count).hasValue(0);
}

La única solución para admitir los dos escenarios es evitar la detección del último spliterator en transmisiones paralelas sin formato. La consecuencia es que la solución realizará operaciones en todos los elementos, pero siempre dará la respuesta correcta.

Tenga en cuenta que en las secuencias secuencial, de todos modos realizará operaciones en todos los elementos.

public static <T> Optional<T> last(Stream<? extends T> stream) {
    Objects.requireNonNull(stream, "stream");

    Spliterator<? extends T> spliterator = stream.spliterator();

    // Find the last spliterator with estimate size (sized parallel streams)
    if(spliterator.hasCharacteristics(Spliterator.SIZED|Spliterator.SUBSIZED)) {
        // Find the last spliterator on sized streams (parallel streams)
        for (Spliterator<? extends T> prev = spliterator.trySplit(); prev != null; prev = spliterator.trySplit()) {
            if (spliterator.getExactSizeIfKnown() == 0) {
                spliterator = prev;
                break;
            }
        }
    }

    // Find the last element of the spliterator
    //AtomicReference<T> last = new AtomicReference<>();
    //spliterator.forEachRemaining(last::set);

    //return Optional.ofNullable(last.get());

    // A better one that supports native parallel streams
    return (Optional<T>) StreamSupport.stream(spliterator, stream.isParallel())
            .reduce((a, b) -> b);
}

Con respecto a la unidad de prueba para esa implementación, las primeras tres pruebas son exactamente las mismas (paralelo secuencial y de tamaño). Las pruebas para el paralelo no dimensionado están aquí:

@Test
@DisplayName("last parallel unsized")
void last_parallel_unsized() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed().parallel();
    stream = StreamSupport.stream(((Iterable<Long>) stream::iterator).spliterator(), stream.isParallel());
    stream = stream.peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(10_000_000L);
}

@Test
@DisplayName("last parallel unsized with skip")
void last_parallel_unsized_with_skip() throws Exception {
    long expected = 10_000_000L;
    AtomicLong count = new AtomicLong();
    Stream<Long> stream = LongStream.rangeClosed(1, expected).boxed().parallel();
    stream = StreamSupport.stream(((Iterable<Long>) stream::iterator).spliterator(), stream.isParallel());
    stream = stream.skip(50_000).peek(num -> count.getAndIncrement());

    assertThat(Streams.last(stream)).hasValue(expected);
    assertThat(count).hasValue(9_950_000L);
}