Потоки Java: как сделать эффективную "отличную и сортировку"?

Предположим, что у меня есть Stream<T> и вы хотите получить только отдельные элементы и отсортированы.

Наивный подход состоял бы в следующем:

Stream.of(...)
    .sorted()
    .distinct()

или, может быть, наоборот:

Stream.of(...)
    .distinct()
    .sorted()

Поскольку реализация обоих из них на самом деле недоступна исходным кодом JDK, я просто задавался вопросом о возможном потреблении памяти и производительности.

Или было бы еще эффективнее написать собственный фильтр следующим образом?

Stream.of(...)
    .sorted()
    .filter(noAdjacentDuplicatesFilter())

public static Predicate<Object> noAdjacentDuplicatesFilter() {
    final Object[] previousValue = {new Object()};

    return value -> {
        final boolean takeValue = !Objects.equals(previousValue[0], value);
        previousValue[0] = value;
        return takeValue;
    };
}

Ответ 1

Когда вы связываете операцию distinct() после sorted(), реализация будет использовать отсортированный характер данных и избежать создания внутреннего HashSet, что может быть продемонстрировано следующей программой

public class DistinctAndSort {
    static int COMPARE, EQUALS, HASHCODE;
    static class Tracker implements Comparable<Tracker> {
        static int SERIAL;
        int id;
        Tracker() {
            id=SERIAL++/2;
        }
        public int compareTo(Tracker o) {
            COMPARE++;
            return Integer.compare(id, o.id);
        }
        public int hashCode() {
            HASHCODE++;
            return id;
        }
        public boolean equals(Object obj) {
            EQUALS++;
            return super.equals(obj);
        }
    }
    public static void main(String[] args) {
        System.out.println("adjacent sorted() and distinct()");
        Stream.generate(Tracker::new).limit(100)
              .sorted().distinct()
              .forEachOrdered(o -> {});
        System.out.printf("compareTo: %d, EQUALS: %d, HASHCODE: %d%n",
                          COMPARE, EQUALS, HASHCODE);
        COMPARE=EQUALS=HASHCODE=0;
        System.out.println("now with intermediate operation");
        Stream.generate(Tracker::new).limit(100)
            .sorted().map(x -> x).distinct()
            .forEachOrdered(o -> {});
        System.out.printf("compareTo: %d, EQUALS: %d, HASHCODE: %d%n",
                          COMPARE, EQUALS, HASHCODE);
    }
}

который будет печатать

adjacent sorted() and distinct()
compareTo: 99, EQUALS: 99, HASHCODE: 0
now with intermediate operation
compareTo: 99, EQUALS: 100, HASHCODE: 200

Промежуточная операция, простая как map(x -> x), не может быть распознана реализацией Stream, следовательно, она должна предполагать, что элементы могут не сортироваться относительно результата функций сопоставления.

Нет никакой гарантии, что такая оптимизация произойдет, однако разумно предположить, что разработчики реализации Stream не удалят эту оптимизацию и даже попытаются добавить больше оптимизаций, поэтому развертывание собственной реализации предотвратит ваш код от использования будущих оптимизаций.

Кроме того, то, что вы создали, - это "предикат с состоянием", который сильно обескуражен и, разумеется, будет разбит при использовании с параллельным потоком.

Если вы не доверяете Stream API для эффективного выполнения этой операции, вам может быть лучше реализовать эту конкретную операцию без Stream API.

Ответ 2

Отказ от ответственности: я знаю, что тестирование производительности сложно и особенно на JVM с необходимыми разминками и контролируемой средой без каких-либо других процессов.

Если я тестирую его, я получаю эти результаты, поэтому кажется, что ваша реализация обеспечивает параллельное выполнение. (Запуск на i7 с 4 ядрами + гиперпоточность).

Итак, ".distinct().sorted()" кажется медленнее. Как предсказано/объяснено Хольгером

Round 1 (Warm up?)
3938
2449
5747
Round 2
2834
2620
3984
Round 3 Parallel
831
4343
6346
Round 4 Parallel
825
3309
6339

Использование кода:

package test.test;

import java.util.Collections;
import java.util.List;
import java.util.Objects;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class SortDistinctTest {

    public static void main(String[] args) {
        IntStream range = IntStream.range(0, 6_000_000);
        List<Integer> collect = range.boxed().collect(Collectors.toList());
        Collections.shuffle(collect);

        long start = System.currentTimeMillis();

        System.out.println("Round 1 (Warm up?)");
        collect.stream().sorted().filter(noAdjacentDuplicatesFilter()).collect(Collectors.counting());
        long fst = System.currentTimeMillis();
        System.out.println(fst - start);

        collect.stream().sorted().distinct().collect(Collectors.counting());
        long snd = System.currentTimeMillis();
        System.out.println(snd - fst);

        collect.stream().distinct().sorted().collect(Collectors.counting());
        long end = System.currentTimeMillis();
        System.out.println(end - snd);

        System.out.println("Round 2");
        collect.stream().sorted().filter(noAdjacentDuplicatesFilter()).collect(Collectors.counting());
        fst = System.currentTimeMillis();
        System.out.println(fst - end);

        collect.stream().sorted().distinct().collect(Collectors.counting());
        snd = System.currentTimeMillis();
        System.out.println(snd - fst);

        collect.stream().distinct().sorted().collect(Collectors.counting());
        end = System.currentTimeMillis();
        System.out.println(end - snd);

        System.out.println("Round 3 Parallel");
        collect.stream().parallel().sorted().filter(noAdjacentDuplicatesFilter()).collect(Collectors.counting());
        fst = System.currentTimeMillis();
        System.out.println(fst - end);

        collect.stream().parallel().sorted().distinct().collect(Collectors.counting());
        snd = System.currentTimeMillis();
        System.out.println(snd - fst);

        collect.stream().parallel().distinct().sorted().collect(Collectors.counting());
        end = System.currentTimeMillis();
        System.out.println(end - snd);

        System.out.println("Round 4 Parallel");
        collect.stream().parallel().sorted().filter(noAdjacentDuplicatesFilter()).collect(Collectors.counting());
        fst = System.currentTimeMillis();
        System.out.println(fst - end);

        collect.stream().parallel().sorted().distinct().collect(Collectors.counting());
        snd = System.currentTimeMillis();
        System.out.println(snd - fst);

        collect.stream().parallel().distinct().sorted().collect(Collectors.counting());
        end = System.currentTimeMillis();
        System.out.println(end - snd);

    }

    public static Predicate<Object> noAdjacentDuplicatesFilter() {
        final Object[] previousValue = { new Object() };

        return value -> {
            final boolean takeValue = !Objects.equals(previousValue[0], value);
            previousValue[0] = value;
            return takeValue;
        };

    }

}