Estou tendo problemas para entender completamente o papel que o combinerdesempenha no reducemétodo Streams .

Por exemplo, o código a seguir não é compilado:

int length = asList("str1", "str2").stream()
            .reduce(0, (accumulatedInt, str) -> accumulatedInt + str.length());

O erro de compilação diz: (incompatibilidade de argumento; int não pode ser convertido em java.lang.String)

mas esse código compila:

int length = asList("str1", "str2").stream()  
    .reduce(0, (accumulatedInt, str ) -> accumulatedInt + str.length(), 
                (accumulatedInt, accumulatedInt2) -> accumulatedInt + accumulatedInt2);

Entendo que o método combinador é usado em fluxos paralelos - portanto, no meu exemplo, ele adiciona duas entradas intermediárias acumuladas.

Mas não entendo por que o primeiro exemplo não é compilado sem o combinador ou como o combinador está resolvendo a conversão de string em int, pois está apenas adicionando duas entradas.

Alguém pode esclarecer isto?

As versões de dois e três argumentos reduceque você tentou usar não aceitam o mesmo tipo para o accumulator.

O argumento dois reduceé definido como :

T reduce(T identity,
         BinaryOperator<T> accumulator)

No seu caso, T é String, portanto, BinaryOperator<T>deve aceitar dois argumentos de String e retornar uma String. Mas você passa para ele um int e um String, o que resulta no erro de compilação que você obteve - argument mismatch; int cannot be converted to java.lang.String. Na verdade, acho que passar 0 como o valor da identidade também está errado aqui, pois uma String é esperada (T).

Observe também que esta versão do reduz processa um fluxo de Ts e retorna um T, portanto você não pode usá-lo para reduzir um fluxo de String a um int.

O argumento três reduceé definido como :

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

No seu caso, U é Inteiro e T é String, portanto, este método reduz um fluxo de String para um Inteiro.

Para o BiFunction<U,? super T,U>acumulador, você pode passar parâmetros de dois tipos diferentes (U e? Super T), que no seu caso são Inteiro e String. Além disso, o valor da identidade U aceita um número inteiro no seu caso, portanto, passar 0 é bom.

Outra maneira de conseguir o que deseja:

int length = asList("str1", "str2").stream().mapToInt (s -> s.length())
            .reduce(0, (accumulatedInt, len) -> accumulatedInt + len);

Aqui, o tipo de fluxo corresponde ao tipo de retorno de reduce, para que você possa usar a versão de dois parâmetros reduce.

Claro que você não precisa usar reducenada:

int length = asList("str1", "str2").stream().mapToInt (s -> s.length())
            .sum();

A resposta de Eran descreveu as diferenças entre as versões de dois e três argumentos reduceem que a primeira se reduz Stream<T>a Tenquanto a segunda se reduz Stream<T>a U. No entanto, ele realmente não explicou a necessidade da função combinadora adicional ao reduzir Stream<T>para U.

Um dos princípios de design da API do Streams é que a API não deve diferir entre fluxos sequenciais e paralelos ou, dito de outra forma, uma API específica não deve impedir que um fluxo seja executado corretamente sequencialmente ou em paralelo. Se suas lambdas tiverem as propriedades corretas (associativas, não interferentes, etc.), um fluxo executado sequencialmente ou em paralelo deverá fornecer os mesmos resultados.

Vamos primeiro considerar a versão de redução de dois argumentos:

T reduce(I, (T, T) -> T)

A implementação seqüencial é direta. O valor da identidade Ié "acumulado" com o elemento de fluxo zeroth para fornecer um resultado. Esse resultado é acumulado com o primeiro elemento de fluxo para fornecer outro resultado, que por sua vez é acumulado com o segundo elemento de fluxo e assim por diante. Depois que o último elemento é acumulado, o resultado final é retornado.

A implementação paralela começa dividindo o fluxo em segmentos. Cada segmento é processado por seu próprio encadeamento da maneira sequencial descrita acima. Agora, se tivermos N threads, teremos N resultados intermediários. Estes precisam ser reduzidos a um resultado. Como cada resultado intermediário é do tipo T e temos vários, podemos usar a mesma função acumuladora para reduzir esses N resultados intermediários para um único resultado.

Agora vamos considerar uma operação hipotética de redução de dois argumentos que reduz Stream<T>a U. Em outros idiomas, isso é chamado de operação "fold" ou "fold-left", e é assim que chamarei aqui. Observe que isso não existe em Java.

U foldLeft(I, (U, T) -> U)

(Observe que o valor da identidade Ié do tipo U.)

A versão sequencial de foldLefté exatamente reduceigual à versão seqüencial, exceto que os valores intermediários são do tipo U em vez do tipo T. Mas, caso contrário, é o mesmo. (Uma foldRightoperação hipotética seria semelhante, exceto que as operações seriam executadas da direita para a esquerda em vez de da esquerda para a direita.)

Agora considere a versão paralela de foldLeft. Vamos começar dividindo o fluxo em segmentos. Podemos então fazer com que cada um dos N threads reduza os valores T em seu segmento em N valores intermediários do tipo U. E agora? Como chegamos de N valores do tipo U a um único resultado do tipo U?

O que está faltando é outra função que combina os vários resultados intermediários do tipo U em um único resultado do tipo U. Se tivermos uma função que combine dois valores de U em um, isso é suficiente para reduzir qualquer número de valores para um - assim como a redução original acima. Portanto, a operação de redução que resulta em um tipo diferente precisa de duas funções:

U reduce(I, (U, T) -> U, (U, U) -> U)

Ou, usando a sintaxe Java:

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

Em resumo, para fazer uma redução paralela a um tipo de resultado diferente, precisamos de duas funções: uma que acumule elementos T para valores U intermediários e uma segunda que combine os valores U intermediários em um único resultado U. Se não estamos trocando de tipo, acontece que a função acumuladora é a mesma que a função combinadora. É por isso que a redução para o mesmo tipo possui apenas a função de acumulador e a redução para um tipo diferente requer funções separadas de acumulador e combinador.

Finalmente, Java não fornece foldLefte foldRightoperações porque implicam uma ordem particular de operações que é inerentemente sequencial. Isso entra em conflito com o princípio de design declarado acima, ao fornecer APIs que suportam igualmente a operação sequencial e paralela.

Desde que eu gosto de rabiscos e flechas para esclarecer conceitos ... vamos começar!

De String para String (fluxo sequencial)

Suponha que você tenha 4 strings: seu objetivo é concatená-las em uma. Você basicamente começa com um tipo e termina com o mesmo tipo.

Você pode conseguir isso com

String res = Arrays.asList("one", "two","three","four")
        .stream()
        .reduce("",
                (accumulatedStr, str) -> accumulatedStr + str);  //accumulator

e isso ajuda você a visualizar o que está acontecendo:

A função acumulador converte, passo a passo, os elementos no seu fluxo (vermelho) no valor final reduzido (verde). A função acumuladora simplesmente transforma um Stringobjeto em outro String.

De String para int (fluxo paralelo)

Suponha que você tenha as mesmas quatro strings: seu novo objetivo é somar seus comprimentos e você deseja paralelizar seu stream.

O que você precisa é algo como isto:

int length = Arrays.asList("one", "two","three","four")
        .parallelStream()
        .reduce(0,
                (accumulatedInt, str) -> accumulatedInt + str.length(),                 //accumulator
                (accumulatedInt, accumulatedInt2) -> accumulatedInt + accumulatedInt2); //combiner

e este é um esquema do que está acontecendo

Aqui, a função acumulador (a BiFunction) permite transformar seus Stringdados em intdados. Sendo o fluxo paralelo, ele é dividido em duas partes (vermelhas), cada uma das quais elaborada independentemente uma da outra e produz apenas o mesmo resultado parcial (laranja). É necessário definir um combinador para fornecer uma regra para mesclar intresultados parciais no final (verde) int.

De String para int (fluxo sequencial)

E se você não quiser paralelizar seu fluxo? Bem, um combinador precisa ser fornecido de qualquer maneira, mas nunca será invocado, uma vez que nenhum resultado parcial será produzido.

Não existe reduzir versão que leva dois tipos diferentes sem um combinador , uma vez que não podem ser executadas em paralelo (não tenho certeza por que este é um requisito). O fato de o acumulador precisar ser associativo torna essa interface praticamente inútil, pois:

list.stream().reduce(identity,
                     accumulator,
                     combiner);

Produz os mesmos resultados que:

list.stream().map(i -> accumulator(identity, i))
             .reduce(identity,
                     combiner);