Объединить значения с потоком Java8

Question

Если у меня есть список с целыми числами, есть ли способ построить другой список, где целые числа суммируются, если разница с заголовком нового списка ниже трэшольда? Я хотел бы решить это, используя потоки Java 8. Он должен работать аналогично Scan operator RxJava.

Example: 5, 2, 2, 5, 13  
Threashold: 2  
Result: 5, 9, 13 

Intermediate results:  
5 
5, 2 
5, 4 (2 and 2 summed) 
5, 9 (4 and 5 summed) 
5, 9, 13 

Answer 1

Последовательное решение поток может выглядеть следующим образом:

List<Integer> result = Stream.of(5, 2, 2, 5, 13).collect(ArrayList::new, (list, n) -> { 
    if(!list.isEmpty() && Math.abs(list.get(list.size()-1)-n) < 2) 
     list.set(list.size()-1, list.get(list.size()-1)+n); 
    else 
     list.add(n); 
}, (l1, l2) -> {throw new UnsupportedOperationException();}); 
System.out.println(result); 

Хотя это выглядит не намного лучше, так как старый хорошее решение:

List<Integer> input = Arrays.asList(5, 2, 2, 5, 13); 
List<Integer> list = new ArrayList<>(); 
for(Integer n : input) { 
    if(!list.isEmpty() && Math.abs(list.get(list.size()-1)-n) < 2) 
     list.set(list.size()-1, list.get(list.size()-1)+n); 
    else 
     list.add(n); 
} 
System.out.println(list); 

кажется, что ваша проблема не ассоциативно, так это не могут быть легко распараллелены. Например, если вы разделите вход на две группы, такие как (5, 2), (2, 5, 13), вы не можете сказать, должны ли быть объединены первые два элемента второй группы до тех пор, пока первая группа не будет обработана. Таким образом, я не могу указать правильную функцию объединителя.

Answer 2

Java 8 способ определить пользовательские IntSpliterator класс:

static class IntThreasholdSpliterator extends Spliterators.AbstractIntSpliterator { 
    private PrimitiveIterator.OfInt it; 
    private int threashold; 
    private int sum; 

    public IntThreasholdSpliterator(int threashold, IntStream stream, long est) { 
     super(est, ORDERED); 
     this.it = stream.iterator(); 
     this.threashold = threashold; 
    } 

    @Override 
    public boolean tryAdvance(IntConsumer action) { 
     if(!it.hasNext()){ 
      return false; 
     } 
     int next = it.nextInt(); 
     if(next<threashold){ 
      sum += next; 
     }else { 
      action.accept(next + sum); 
      sum = 0; 
     } 
     return true; 
    } 

} 

public static void main(String[] args) 
{ 
    IntThreasholdSpliterator s = new IntThreasholdSpliterator(3, IntStream.of(5, 2, 2, 5, 13), 5); 
    List<Integer> rs= StreamSupport.intStream(s, false).mapToObj(Integer::valueOf).collect(toList()); 
    System.out.println(rs); 
} 

Также вы можете взломать его как

List<Integer> list = Arrays.asList(5, 2, 2, 5, 13); 
    int[] sum = {0}; 
    list = list.stream().filter(s -> { 
     if(s<=2) sum[0]+=s; 
     return s>2; 
    }).map(s -> { 
     int rs = s + sum[0]; 
     sum[0] = 0; 
     return rs; 
    }).collect(toList()); 
    System.out.println(list); 

Но я не уверен, что это хак хорошая идея для производства кода.

Answer 3

Как Tagir Valeev observed, (+1) функция объединения не является ассоциативной, поэтому reduce() не будет работать, и невозможно создать функцию объединителя для Collector. Вместо этого эту функцию объединения нужно применять слева направо, при этом предыдущий частичный результат будет передан в следующую операцию. Это называется операцией fold-left, и, к сожалению, Java-потоки не имеют такой операции.

(Должен ли они? Дай мне знать.)

Можно сортировать-вписать свою откидывающуюся влево операцию с forEachOrdered во время захвата и мутирует объект для проведения частичного состояния. Во-первых, давайте извлекать функцию объединения в его собственный метод:

// extracted from Tagir Valeev's answer 
void combine(List<Integer> list, int n) { 
    if (!list.isEmpty() && Math.abs(list.get(list.size()-1)-n) < 2) 
     list.set(list.size()-1, list.get(list.size()-1)+n); 
    else 
     list.add(n); 
} 

Затем создайте первоначальный список результатов и вызова функции объединения изнутри forEachOrdered:

List<Integer> result = new ArrayList<>(); 
IntStream.of(5, 2, 2, 5, 13) 
     .forEachOrdered(n -> combine(result, n)); 

Это дает желаемый результат

[5, 9, 13] 

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

Answer 4

Я знаю, что мастера Stream «Tagir Valeev» и «Stuart Marks» уже указывали, что reduce() не будет работать, потому что функция объединения не является ассоциативной, и я рискую здесь несколькими downvotes. Во всяком случае:

Как насчет того, чтобы заставить поток быть последовательным? Разве мы не сможем использовать сокращение?Разве это свойство ассоциативности не требуется только при использовании параллелизма?

Stream<Integer> s = Stream.of(5, 2, 2, 5, 13); 
    LinkedList<Integer> result = s.sequential().reduce(new LinkedList<Integer>(), 
       (list, el) -> { 
        if (list.isEmpty() || Math.abs(list.getLast() - el) >= 2) { 
         list.add(el); 
        } else { 
         list.set(list.size() - 1, list.getLast() + el); 
        } 
        return list; 
       }, (list1, list2) -> { 
         //don't really needed, as we are sequential 
         list1.addAll(list2); return list1; 
         });