Фильтр с отслеживанием состояния для упорядоченного потока

Question

У меня проблема, и мне интересно, есть ли решение с использованием Streams.

Представьте, что у вас есть упорядоченный поток Объектов;давайте предположим, что поток целых чисел.

 Stream<Integer> stream = Stream.of(2,20,18,17,4,11,13,6,3,19,4,10,13....)

Теперь я хочу отфильтровать все значения, где разница значения и предыдущего числа перед этим значением больше, чем n .

stream.filter(magicalVoodoo(5))
// 2, 20, 4, 11, 3, 19, 4, 10 ...

Есть ли возможность сделать это?

Answer 1

Да, это возможно, но для сравнения вам понадобится предикат с отслеживанием состояния, который отслеживает предыдущее значение.Это означает, что его можно использовать только для последовательных потоков: с параллельными потоками вы столкнетесь с условиями гонки.

К счастью, большинство потоков по умолчанию являются последовательными, но если вам нужно сделать это для потоков из неизвестного источника, вы можете проверить с помощью isParallel() и либо выбросить исключение, либо преобразовать его в последовательный поток.используя sequential().

Пример:

public class DistanceFilter implements IntPredicate {

    private final int distance;
    private int previousValue;

    public DistanceFilter(int distance) {
        this(distance, 0);
    }

    public DistanceFilter(int distance, int startValue) {
        this.distance = distance;
        this.previousValue = startValue;
    }

    @Override
    public boolean test(int value) {
        if (Math.abs(previousValue - value) > distance) {
            previousValue = value;
            return true;
        }
        return false;
    }

    // Just for simple demonstration
    public static void main(String[] args) {
        int[] ints = IntStream.of(2, 20, 18, 17, 4, 11, 13, 6, 3, 19, 4, 10, 13)
                .filter(new DistanceFilter(5))
                .toArray();

        System.out.println(Arrays.toString(ints));
    }
}

Я использовал IntStream здесь, потому что это лучший тип для этого, но концепция была бы аналогичной для Stream<Integer> (или другие типы объектов).

Answer 2

Потоки не предназначены для такого рода задач.Я бы использовал другой способ для достижения этой цели, который не использует потоки.Но, если вам действительно нужно использовать потоки, решение должно обойти определенные ограничения из-за конструкции потоков и лямбд, и поэтому выглядит довольно странно:

int[] previous = new int[1];
previous[0] = firstElement;
... = stream.filter(n -> {
        boolean isAllowed = (abs(n - previous[0]) > 5);
        if (isAllowed)
            previous[0] = n;
        return isAllowed;})

Обратите внимание, что переменная previous является единицейэлемент массиваЭто хак из-за того, что лямбда не может изменять переменные (он может изменять элемент массива, но не сам массив).