Ten wpis jest kontynuacją artykułu o strumieniach.

Operacje redukcji

Operacja redukcji (składania) pobiera sekwencję elementów wejściowych i składa je w pojedynczy wynik przez powtarzanie pewnych operacji. Przykładem operacji redukcji jest sumowanie elementów czy utworzenie listy elementów. Klasy strumieni wykorzystują metody reduce(), collect() podobnie jak wyspecjalizowane metody sum(), max(), czy count().

w praktyce programistycznej takie operacje jak operacja sumowania wykonuje się w prostej pętli sekwencyjnej:

int sum = 0;
for(int x: values:numbers {
sum += x;
}

Istnieje wiele przyczyn dla których należy przedkładać operacje redukcji ponad akumulację mutatywną, taką jak pokazana powyżej. Operacja redukcji operuje na strumieniu jako całości,a nie na pojedynczych elementach i jeśli tylko funkcje działające na strumieniu są asocjacyjne i bezstanowe to właściwie skonstruowana operacja redukcji jest zawsze zdolna do paralelizacji.

long sum = values.stream().reduce(0, (x,y) -> x + y);

Wystarczy stream() zamienić na parallelStream(), a operacja może przebiegać a trybie równoległym.

Redukcja pozwala na łatwą paralelizację ponieważ implementacja zapewnia wykonywanie operacji równoległych na podzbiorach, a następnie na odpowiednie połączenie wyników pośrednich w końcowy, poprawny wynik.

W trybie akumulacji mutatywnej należy zapewnić synchronizację, więc jeśli nawet istnieje możliwość paralelizacji to zastosowanie synchronizacji eliminuje wszystkie korzyści z równoległości.

Operacja reduce() zapewnia korzyści z paralelizacji bez obciążeń powodowanych przez konieczność synchronizacji zmiennych.

Operacja redukcji na elementach typu <T> dająca wynik typu <U> wymaga trzech parametrów: identity, accumulator,combiner.

<U> U reduce(U identity, BiFunction<U, ? super T, U> accumulator, BinaryOperator<U> combiner);

Identyczność (identity) jest zarówno wartością początkową ziarna dla redukcji jak i domyślnym wynikiem, jeśli nie ma elementów wejściowych.

Funkcja akumulacyjna (accumulator) pobiera częściowy wynik oraz następny wynik i wytwarza nowy częściowy wynik.

Funkcja kombinacyjna (łącząca) (combiner) łączy dwa wyniki częściowe i tworzy nowy wynik częściowy. Funkcja kombinacyjna jest konieczna przy redukcji równoległej, gdyż wtedy wejście jest partycjonowane, obliczana jest częściowa akumulacja dla każdej partycji, a następnie częściowe wyniki są łączone do wytworzenia wyniku końcowego.

Bardziej formalnie, identyczność musi być identycznością dla funkcji łączącej. Jeżeli np. jest to suma to identyczność będzie 0, gdyż 0 nie zmienia wyniku przy sumowaniu. Jeżeli będzie to mnożenie czy dzielenie – będzie to 1, gdyż 1 nie zmienia wyniku przy mnożeniu i dzieleniu.

Dodatkowo funkcja łącząca musi być assocjacyjna i kompatybilna z funkcją akumulatorową.

Redukcja mutatywna

Operacja redukcji mutatywnej umieszcza elementy wejściowe w mutowalnym kontenerze takim jak Collection albo StringBuilder w trakcie przetwarzania elementów w strumieniu. Jeśli mamy strumień stringów i chcemy połączyć je w jeden łańcuch możemy uczynić to za pomocą operacji redukcji (streams2.Reduction01.java):

String concatenated = Arrays.stream(values).reduce("", String::concat);

Oczywiście otrzymamy to co chcemy, a operacja może nawet przebiegać równolegle. Problemem jednak jest wydajność operacji. Nieco wydajniej można tę operację przeprowadzić używając StringBuilder. Wtedy lepiej jest użyć operacji collect().

Aby umieścić strumień elementów w ArrayList nie możemy tego zrobić przez dodawanie elementów ze strumienia w trakcie iteracji po strumieniu. przy użyciu pętli forEach. Możemy użyć operacji collect() (streams2.Reduction02.java):

Stream<String> stream = Arrays.stream(values);
ArrayList<String> strings = stream.collect(() -> new
   ArrayList<>(), (c, e) -> c.add(e), (c1, c2) ->
c1.addAll(c2));

Zamiennie można wyciągnąć operację mapowania z funkcji akumulatorowej i zapisać operację nieco zwięźlej (streams2.Reduction03.java):

Stream<String> stream = Arrays.stream(values);
List<String> strings1 = stream.map(Object::toString).collect(
	ArrayList::new, ArrayList::add, ArrayList::addAll);

Zamiennie można użyć abstrakcji Collector to uchwycenia wszystkich trzech aspektów (supplier, accumulator, combiner) za jednych zamachem streams2.Reduction04.java):

Stream<String> stream = Arrays.stream(values);
List<String> strings1 = 
   stream.map(Object::toString).collect(Collectors.toList());

Pakowanie mutowalnych redukcji w Collector oprócz uproszczenia składni przynosi jeszcze jedną korzyść: kompozycyjność. Klasa Collector zawiera liczne predefiniowane fabryki kolektorów, włączając kombinatory, które przekształcają jeden kolektor w drugi.

Przypuśćmy, że mamy kolektor obliczający sumę pensji w strumieniu pracowników:

Collector<Pracownik,?, Integer> sumowaniePensji = 
    Collectors.summingInt(Pracownik::getPensja);

Znak zapytania oznacza, że nie jest dla nas istotne jakich pośrednich reprezentacji użyje wewnętrznie kolektor.

Możemy go wykorzystać tak (streams2.Reduction05.java):

Integer aa = str.collect(Collectors.summingInt(Pracownik::getPensja));
System.out.println(aa);

albo tak:

Integer ab = str.collect(sumowaniePensji);	

W przypadku, gdy chcemy utworzyć kolektor grupujący pensje według departamentu może ponownie użyć kolektora z jednoczesnym użyciem groupingBy (streams2.Reduction06.java):

Map<String, Integer> salByDept = 
   str.collect(Collectors.groupingBy(
Pracownik::getDepartament, sumowaniePensji));

Redukcja konkurencyjna (współbieżna)

Wykonywanie niektórych złożonych operacji redukcji równolegle, np. collect(), w wyniku których powstają nowe kolekcje mogą być bardzo kosztowne czasowo. Dlatego na ogół równolegle wykonuje się
operacje przy użyciu kolekcji współbieżnych

Redukcja konkurencyjna to redukcja równoległa z użyciem kolekcji współbieżnych np. ConcurrentHashMap.

Collector, który dostarcza redukcji konkurencyjnej jest oznakowany jako Collector.Characteristics.CONCURRENT. Niestety kolekcje współbieżne mają swoją ciemną stronę. Jeśli wiele wątków składa wyniki współbieżnie w dzielonym kontenerze, porządek w którym wyniki są układane nie jest deterministyczny. W konsekwencji redukcja konkurencyjna jest możliwa tylko wtedy, gdy uporządkowanie nie jest istotne dla przetwarzanego strumienia.

Implementacja Stream.collect(Collector) będzie wykonywała redukcję współbieżnie jeŚli:

  • Strumień jest równoległy
  • Kolektor ma charakterystykę Collector.Characteristics.CONCURRENT i
  • albo strumień jest nieuporządkowany albo kolektor ma charakterystykę Collector.Characteristics.UNORDERED

Używając metody BaseStream.unordered() można stwierdzić czy strumień jest uporządkowany.

W przypadku, gdy uporządkowanie elementów jest istotne nie można używać redukcji współbieżnej.

Asocjacyjność

Operator lub funkcja jest asocjacyjna gdy spęlnia warunek:

(a op b) op c == a op (b op c)

gdzie ‘op’ jest operacją np. mnożenia.

W równoległości wykorzystuje się szczególnie asocjacyjność typu:

a op b op c op d == (a op b) op (c op d)

Nispokopoziomowa konstrukcja strumieni

Metody stream klasy Collection i Arrays pozwalają na otrzymywanie strumieni. Klasa StreamSupport zawiera kilka niskopoziomowych metod do tworzenia strumieni.

Spliterator jest równoległym analogiem iteratora Iterator. Spliterator opisuje (możliwie nieskończoną) kolekcję elementów z możliwością sekwencyjnego przechodzenia naprzód, bąbelkowego przechodzenia i dzielenia niektórych porcji wejścia na spliteratory, które mogą być przetwarzane równolegle. Na najniższym poziomie, wszystkie strumienie są kierowane przez spliterator.

Istnieją różne możliwości implementacji spliteratorów, z których niemal wszystkie są kompromisem pomiędzy prostotą implementacji, a ich efektywnością.

Najprostszym spliteratorem, ale najmniej efektywnym jest spliterator utworzony z iteratora przy użyciu metody Spliterators.spliteratorUnknownSize(java.util.Iterator, int).

Lepszy spliterator będzie dostarczał dobrze zbalansowanych podziałów o dobrze znanej wielkości oraz licznych charakterystyk spliteratora lub danych, które mogą być użyte przez implementację do optymalizacji wykonania.

Spliteratory dla mutowalnych danych podlegają dodatkowym wyzwaniom: taktowanie wiązania do danych ponieważ dane mogą zmieniać się pomiędzy czasem, gdy spliterator jest tworzony, a czasem gdy potok strumienia jest wykonywany.

Idealnie, spliterator powinien mieć w charakterystyce IMMUTABLE lub CONCURRENT, a jeśli nie ma powinien zapewniać późne wiązanie.

Jeśli źródło danych nie może bezpośrednio dostarczyć rekomendowanego spliteratora, może pośrednio dostarczyć spliteratora używając Supplier i skonstruować strumień przez wersje stream(), które akceptują Supplier. Spliterator jest otrzymywany z dostarczyciela tylko wtedy, gdy zacznie się kończąca operacja w potoku strumienia.

Dodaj komentarz

Twój adres email nie zostanie opublikowany. Pola, których wypełnienie jest wymagane, są oznaczone symbolem *