Java Streams - Фильтрация ранее отфильтрованных значений

Question

Я экспериментирую с потоками Java и пытаюсь выяснить, что возможно, а также их сильные и слабые стороны. В настоящее время я пытаюсь реализовать Сито Эратосфена с использованием потока, но, похоже, не может найти хороший способ перебрать ранее отфильтрованные значения, не сохраняя их в отдельной коллекции.

Я хотел сделать что-то вроде этого:

IntStream myStream = IntStream.range(0,3); 
myStream.filter(s -> { 
    System.out.print("[filtering "+s+"] "); 
    myStream.forEach(q -> System.out.print(q+", ")); 
    System.out.println(); 
    return true; //eventually respond to values observed on the line above 
}); 

С желаемым выходом:

[filtering 0] 
[filtering 1] 0, 
[filtering 2] 0, 1, 
[filtering 3] 0, 1, 2, 

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

Однако приведенный выше пример дает мне ошибку в NetBeans:

local variables referenced from a lambda expression must be final or effectively final 

Это, кажется, потому что я ссылки myStream в фильтр, который уже действует на myStream. Есть ли хороший способ обойти эту ошибку (т. Е. Сделать окончательную копию потока, содержащего только те значения, которые были отфильтрованы до сих пор), или есть лучший подход к этой проблеме без использования отдельной коллекции для хранения значения?

Answer 1

Невозможно обработать поток более одного раза, поэтому вызов метода myStream.forEach внутри метода фильтра невозможен.

Внутри фильтра можно создать новый IntStream.

Обратите внимание, что вам придется добавить некоторые операции терминала к наружному трубопроводу поток для того, чтобы быть обработаны:

IntStream myStream = IntStream.range(0,4); 
myStream.filter(s -> { 
    System.out.print("[filtering "+s+"] "); 
    IntStream.range(0,s).forEach(q -> System.out.print(q+", ")); 
    System.out.println(); 
    return true; //eventually respond to values observed on the line above 
}).forEach(i->{}); 

Это дает:

[filtering 0] 
[filtering 1] 0, 
[filtering 2] 0, 1, 
[filtering 3] 0, 1, 2, 

Answer 2

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

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

Листинг простых чисел с использованием потоков (с помощью решета Эратосфена):

List<Integer> primes = new ArrayList<Integer>(); 

IntStream.iterate(2, i -> i + 1) 
    .limit(UPPER_BOUND) 
    .filter(i -> { 
     for(int j=0; j<primes.size(); j++) { 
      int prime = primes.get(j); 

      if(prime > Math.sqrt(i)) { 
       break; 
      } 

      if(i % prime == 0) { 
       return false; 
      } 
     } 
     return true; 
    }) 
    .forEach(primes::add); 

Традиционный, что эквивалентно, подход без использования потоков:

List<Integer> primes = new ArrayList<Integer>(); 

for(int i=2; i < UPPER_BOUND; i++) { 
    boolean isPrime = true; 

    for(int j=0; j<primes.size(); j++) { 
     int prime = primes.get(j); 

     if(prime > Math.sqrt(i)) { 
      break; 
     } 

     if(i % prime == 0) { 
      isPrime = false; 
      break; 
     } 
    } 

    if(isPrime) { 
     primes.add(i); 
    } 
} 

Сравнение производительности:

я экспериментировал с каждой функцией, последовательно демонстрируя, что традиционный подход на самом деле быстрее, чем использование потоков в этом случае. Подход потоков последовательно принимал 1.5 раз дольше, чтобы найти все простые числа менее одного миллиона по сравнению с традиционным подходом (в среднем 106 мс и 70 мс соответственно на моей машине).

Это различие в производительности может быть легко выполнено, если функция .parallel() потока может позволить легко распараллелить проблему. Однако в этом случае распараллеливание непросто, поскольку ArrayList не является потокобезопасным и быстро приведет к ошибкам и/или неточным результатам.

Вывод:

Если предположить, что другие ответы являются правильными, фильтрующие уже отфильтрованные данные в фильтре на том же самом потоке не представляется возможным в Java.

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

Answer 3

Мне удалось создать бесконечные Stream простых чисел, используя Сито Эратосфена, но на самом деле оно не использует прошлые значения. Вместо этого он удаляет кратные штриха в хвосте (ленивым путем, потому что хвост бесконечен), как и исходное сито алгоритма Эратосфена. Для этого я использовал Iterator как вспомогательный (потому что Stream можно использовать только один раз) и реализовал lazyConcat для потоков.

class StreamUtils { 
    public static IntStream fromIterator(PrimitiveIterator.OfInt it) { 
     return StreamSupport.intStream(
       Spliterators.spliteratorUnknownSize(it, Spliterator.ORDERED), false); 
    } 

    public static IntStream lazyConcat(Supplier<IntStream> a, Supplier<IntStream> b) { 
     return StreamSupport.intStream(new Spliterator.OfInt() { 
      boolean beforeSplit = true; 
      Spliterator.OfInt spliterator; 

      @Override 
      public OfInt trySplit() { 
       return null; 
      } 

      @Override 
      public long estimateSize() { 
       return Long.MAX_VALUE; 
      } 

      @Override 
      public int characteristics() { 
       return Spliterator.ORDERED; 
      } 

      @Override 
      public boolean tryAdvance(IntConsumer action) { 
       boolean hasNext; 
       if (spliterator == null) { 
        spliterator = a.get().spliterator(); 
       } 
       hasNext = spliterator.tryAdvance(action); 
       if (!hasNext && beforeSplit) { 
        beforeSplit = false; 
        spliterator = b.get().spliterator(); 
        hasNext = spliterator.tryAdvance(action); 
       } 
       return hasNext; 
      } 
     }, false); 
    } 
} 

Мой Решето Эратосфена поток выглядит следующим образом:

class Primes { 
    public static IntStream stream() { 
     return sieve(IntStream.iterate(2, n -> n + 1)); 
    } 

    private static IntStream sieve(IntStream s) { 
     PrimitiveIterator.OfInt it = s.iterator(); 
     int head = it.nextInt(); 
     IntStream tail = StreamUtils.fromIterator(it); 
     return StreamUtils.lazyConcat(
       () -> IntStream.of(head), 
       () -> sieve(tail.filter(n -> n % head != 0))); 
    } 
} 

Тогда мы можем использовать его таким образом:

System.out.println(Primes.stream().limit(20).boxed().collect(Collectors.toList())); 

Выход:

[2, 3 , 5, 7, 11, 13, 17, 19, 23, 29, 31, 37, 41, 43, 47, 53, 59, 61, 67, 71]

Я думаю, что это было отличное упражнение, но, похоже, оно совсем неэффективно и несовместимо с стеком.

Answer 4

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

List<Integer> primes = IntStream.range(2, UPPER_BOUND) 
    .collect(ArrayList::new, 
      (list, number) -> { 
       for(int j=0; j < list.size(); j++) { 
        int prime = list.get(j); 

        if(prime > Math.sqrt(number)) { 
         break; 
        } 

        if(number % prime == 0) { 
         return; 
        } 
       } 

       list.add(number); 
      }, 
      List::addAll); 

ArrayList::new создает новый список, который затем ссылается потребителем как list. Потребитель вызывается для каждого элемента в потоке с элементом number.

List::addAll относится только к параллельным потокам, которые в любом случае не могут использоваться для этого алгоритма.