Strumień obliczeń statycznych: suma skumulowana

Question

Zakładając, że mam aplikację Java IntStream, czy można ją przekonwertować na IntStream z łącznymi sumami? Na przykład strumień rozpoczynający się od [4, 2, 6, ...] powinien zostać przekonwertowany na [4, 6, 12, ...].

Bardziej ogólnie, w jaki sposób należy postępować w przypadku realizacji operacji z użyciem strumienia stanowego? To powinno być możliwe:

Z oczywistym ograniczeniem, że działa to tylko na sekwencyjnych strumieniach. Jednak Stream.map wyraźnie wymaga funkcji mapowania bezstanowego. Czy mam rację, jeśli brakuje operacji Stream.statefulMap lub Stream.cumulative, czy brakuje punktu strumieni Java?

Porównaj na przykład do Haskell, gdzie funkcja scanl1 rozwiązuje dokładnie ten przykład:

scanl1 (+) [1 2 3 4] = [1 3 6 10] 

Answer 1

Możesz to zrobić, używając numeru atomowego. Na przykład:

import java.util.concurrent.atomic.AtomicLong; 
import java.util.stream.IntStream; 
import java.util.stream.LongStream; 

public class Accumulator { 
    public static LongStream toCumulativeSumStream(IntStream ints){ 
     AtomicLong sum = new AtomicLong(0); 
     return ints.sequential().mapToLong(sum::addAndGet); 
    } 

    public static void main(String[] args){ 
     LongStream sums = Accumulator.toCumulativeSumStream(IntStream.range(1, 5)); 
     sums.forEachOrdered(System.out::println); 
    } 
} 

This Wyjścia:

1 
3 
6 
10 

Użyłem Long do przechowywania sum, bo to całkiem możliwe, że dwa ints sumują się do ponad Integer.MAX_VALUE i długa mniej szansy na przelanie.

Answer 2

Jest to możliwe do zrobienia z kolektorem, który następnie tworzy nowy strumień:

class Accumulator { 
    public static void accept(List<Integer> list, Integer value) { 
     list.add(value + (list.isEmpty() ? 0 : list.get(list.size() - 1))); 
    } 

    public static List<Integer> combine(List<Integer> list1, List<Integer> list2) { 
     int total = list1.get(list1.size() - 1); 
     list2.stream().map(n -> n + total).forEach(list1::add); 
     return list1; 
    } 
} 

ten jest używany jako:

myIntStream.parallel() 
    .collect(ArrayList<Integer>::new, Accumulator::accept, Accumulator::combine) 
    .stream(); 

Mam nadzieję, że widać, że ważną cechą tego Kolektor jest taki, że nawet jeśli strumień jest równoległy, gdy instancje Accumulator są łączone, dostosowuje sumy.

To oczywiście nie jest tak wydajne, jak operacja na mapie, ponieważ zbiera cały strumień, a następnie tworzy nowy strumień. Ale to nie jest tylko szczegół implementacji: jest to niezbędna funkcja faktu, że strumienie mają być potencjalnie przetwarzane jednocześnie.

Przetestowałem to z IntStream.range(0, 10000).parallel() i działa poprawnie.