Bem, você não disse o que você usa como a classe T
e nem o que são essas funções que operam sobre ela para produzir String
s. Então vou inventar uma classe para isso:
class Colorido {
private void complica() {
try {
Thread.sleep((int) (Math.random() * 50));
} catch (InterruptedException e) {
// Ignora.
}
}
public String getVermelho() { complica(); return "vermelho"; }
public String getLaranja() { complica(); return "laranja"; }
public String getAmarelo() { complica(); return "amarelo"; }
public String getLima() { complica(); return "lima"; }
public String getVerde() { complica(); return "verde"; }
public String getCiano() { complica(); return "ciano"; }
public String getAzul() { complica(); return "azul"; }
public String getVioleta() { complica(); return "violeta"; }
public String getRosa() { complica(); return "rosa"; }
public String getRoxo() { complica(); return "roxo"; }
public String getBranco() { complica(); return "branco"; }
public String getPreto() { complica(); return "preto"; }
public String getMarrom() { complica(); return "marrom"; }
public String getOliva() { complica(); return "oliva"; }
public String getBege() { complica(); return "bege"; }
public String getCinza() { complica(); return "cinza"; }
public static List<Function<Colorido, String>> funcoes() {
return Arrays.asList(
Colorido::getVermelho, Colorido::getLaranja, Colorido::getAmarelo, Colorido::getLima,
Colorido::getVerde, Colorido::getCiano, Colorido::getAzul, Colorido::getVioleta,
Colorido::getRosa, Colorido::getRoxo, Colorido::getBranco, Colorido::getPreto,
Colorido::getMarrom, Colorido::getOliva, Colorido::getBege, Colorido::getCinza
);
}
}
Esta é decerto uma classe bem boba. No entanto, observe que o método funcoes
vai devolver uma lista com 16 funções que operam sobre o objeto. Cada função retorna o nome de uma cor diferente, mas todas elas vão demorar algum tempinho para fazer isso graças ao Thread.sleep
de tempo aleatório dentro do método complica
. Afinal de contas, se queremos testar paralelismo, é importante que as threads não executem de forma demasiadamente rápida para evitarmos que uma comece quando a outra já terminou.
Então, vamos testar o map-reduce que você colocou na pergunta:
public static <T> List<String> mapReduceSequencial(
List<Function<T, String>> funcoes, // inicializa com as diversas funções
T objetoSobProcessamento) // parâmetro arbitrário
{
List<String> retornoProcessamento = funcoes
.stream()
.map(f -> f.apply(objetoSobProcessamento))
.reduce(new ArrayList<>(), (listaAcumulada, novoValor) -> {
listaAcumulada.add(novoValor);
return listaAcumulada;
}, (l1, l2) -> {
l1.addAll(l2);
return l1;
});
return Collections.unmodifiableList(retornoProcessamento);
}
Ele é invocado assim:
System.out.println(mapReduceSequencial(Colorido.funcoes(), new Colorido()));
Eis a saída (acrescentei a quebra de linha só para facilitar a leitura):
[vermelho, laranja, amarelo, lima, verde, ciano, azul, violeta,
rosa, roxo, branco, preto, marrom, oliva, bege, cinza]
Observe que a ordem das cores é a mesma da lista.
E se alterarmos aquele .stream()
para .parallelStream()
, o que acontece? Neste caso, o resultado vai variar por causa do Thread.sleep
aleatório e por causa do indeterminismo inerente de ver várias threads simultâneas, mas em um teste que fiz o resultado foi esse:
[verde, rosa, roxo, ciano, azul, branco, violeta, preto,
vermelho, laranja, amarelo, marrom, oliva, lima, verde, rosa,
roxo, ciano, azul, branco, violeta, preto, vermelho, laranja,
amarelo, marrom, oliva, lima, bege, cinza, verde, rosa,
roxo, ciano, azul, branco, violeta, preto, vermelho, laranja,
amarelo, marrom, oliva, lima, verde, rosa, roxo, ciano,
azul, branco, violeta, preto, vermelho, laranja, amarelo, marrom,
oliva, lima, bege, cinza, verde, rosa, roxo, ciano,
azul, branco, violeta, preto, vermelho, laranja, amarelo, marrom,
oliva, lima, verde, rosa, roxo, ciano, azul, branco,
violeta, preto, vermelho, laranja, amarelo, marrom, oliva, lima,
bege, cinza, verde, rosa, roxo, ciano, azul, branco,
violeta, preto, vermelho, laranja, amarelo, marrom, oliva, lima,
verde, rosa, roxo, ciano, azul, branco, violeta, preto,
vermelho, laranja, amarelo, marrom, oliva, lima, bege, cinza]
Bem, esse resultado claramente está muito errado. Isso significa que a implementação do seu reduce
não funciona em Stream
s paralelos (mas tenha calma que daqui a pouco explico como consertar isso). Observe que há várias subsequências se repetindo nesse resultado aí.
Olhando a documentação do método Stream<T>.reduce(U, BiFunction<U, ? super T, U>, BinaryOperator<U>)
cujos parâmetros são chamados respectivamente de identity
, accumulator
e combiner
, encontramos isso:
This is equivalent to:
U result = identity;
for (T element : this stream)
result = accumulator.apply(result, element)
return result;
but is not constrained to execute sequentially.
Traduzindo para o português:
Isto é equivalente a [código] mas não limitado a executar sequencialmente.
Vamos dar uma olhada primeiro nessa linha:
U result = identity;
Isso significa que sua ArrayList
é o primeiro valor de result
.
Agora essa linha dentro do for
:
result = accumulator.apply(result, element)
Isso significa que a partir da lista dada, uma outra lista será produzida e com base nela uma outra lista e assim sucessivamente até os elementos acabarem. Ocorre que no seu caso a lista produzida é sempre a mesma, e portanto ao invés de você produzir novos resultados com base nos resultados parciais anteriores, você está na verdade modificando os resultados parciais anteriores e os devolvendo como se fossem novos.
O resultado é que ao fazer isso em paralelo, quando o combinador for utilizado, você vai ter que l1
e l2
apontam para o mesmo objeto, que é a ArrayList
, e com isso, quando você chama l1.addAll(l2)
, você vai adicionar a lista a ela mesma, o que significa que vai duplicar todos os elementos existentes nela! Junte isso com o fato das threads em paralelo estarem demorando tempos aleatórios/imprevisíveis e o resultado é aquela bagunça que foi produzida.
Voltando lá no javadoc, também está escrito isso:
This means that for all u
, combiner(identity, u)
is equal to u
.
Traduzindo:
Isso significa que para todo u
, combiner(identity, u)
é igual a u
.
Isso significa, de uma forma algébrica, que identity
deve ser o elemento neutro da operação combiner
, assim como 0 é o elemento neutro da soma e 1 é o elemento neutro da multiplicação. Qualquer coisa combinada com identity
deve resultar nessa mesma coisa.
O combiner
é o seu último lambda que faz o addAll
. Em seus dois lambdas você altera a lista, e portanto você altera o que é considerado identity
. Logo, ao invocar combiner(identity, u)
você acaba tendo uma lista que não é igual ao que o u
era antes e sim uma que equivale ao u
dobrado. Isso significa que os seus lambdas violam as regras definidas pela documentação do reduce
ao violar a regra do elemento neutro do combiner
. Assim sendo, é óbvio que coisas ruins vão acontecer.
Podemos alterar algumas coisas nos seus lambdas para restaurar o comportamento/contrato exigido pelo reduce
:
public static <T> List<String> mapReduceParalelo2(
List<Function<T, String>> funcoes, // inicializa com as diversas funções
T objetoSobProcessamento) // parâmetro arbitrário
{
List<String> retornoProcessamento = funcoes
.parallelStream()
.map(f -> f.apply(objetoSobProcessamento))
.reduce(Collections.emptyList(), (listaAcumulada, novoValor) -> {
List<String> novaLista = listaAcumulada.isEmpty() ? new ArrayList<>() : listaAcumulada;
novaLista.add(novoValor);
return novaLista;
}, (l1, l2) -> {
if (l1.isEmpty()) return l2;
if (l2.isEmpty()) return l1;
l1.addAll(l2);
return l1;
});
return Collections.unmodifiableList(retornoProcessamento);
}
O nome do método é mapReduceParalelo2
porque eu vou chamar o que produziu o resultado bagunçado de mapReduceParalelo1
. No método parReduceParalelo2
, note que o identity
é Collections.emptyList()
. Fiz isso para enfatizar que o identity
não deve ser alterado (tentar fazer isso resultaria numa exceção).
O accumulator
testa se a lista dada é um identity
para evitar problemas, criando uma nova se não for. Se a lista dada não for um identity
, então ela pode ser seguramente alterada pois a lista original não seria reutilizada em lugar nenhum se uma nova fosse criada.
Já o combiner
, verifica se uma das listas é um identity
, retornando a outra neste caso. Se nenhuma delas for o identity
, então não há problema em fazer-se o addAll
ao invés de criar uma nova pois nenhuma das listas originais seriam reutilizadas depois.
A moral da história é que o identity
sempre é reutilizado, enquanto os demais resultados parciais não são. Se o identity
fosse sempre recriado ao invés de reutilizado, o mapReduceParalelo1
funcionaria.
Isso pode ter ficado meio complicado porque na verdade o reduce
foi projetado para trabalhar com dados imutáveis, ao menos com um identity
imutável. Embora seja possível usar o reduce
para coisas mutáveis, o resultado é esse troço meio complicado e confuso que saiu. Já o collect
é diferente. O collect
foi de fato projetado para trabalhar com coisas mutáveis. Por exemplo:
public static <T> List<String> mapCollectParalelo(
List<Function<T, String>> funcoes, // inicializa com as diversas funções
T objetoSobProcessamento) // parâmetro arbitrário
{
List<String> retornoProcessamento = funcoes
.parallelStream()
.map(f -> f.apply(objetoSobProcessamento))
.collect(ArrayList::new, List::add, List::addAll);
return Collections.unmodifiableList(retornoProcessamento);
}
Observe a simplicidade dos parâmetros do método collect
. Observe que todos os três parâmetros são method-references. O primeiro, ao invés de ser um objeto fixo para servir de identity
sempre sendo reutilizado, é o Supplier
que é usado para criar novas listas, no caso o construtor de ArrayList
. O segundo parâmetro (accumulator
) é o que adiciona elementos à lista, ou seja o método add
(e o resultado dele é void
, pois a operação esperada aqui é uma mutação). O terceiro parâmetro (combiner
) que combina elementos é o addAll
(também void
). Observe que é usado parallelStream()
. O resultado é o que se espera (e se fosse sequencial, também é):
[vermelho, laranja, amarelo, lima, verde, ciano, azul, violeta,
rosa, roxo, branco, preto, marrom, oliva, bege, cinza]
Talvez você perceba que juntar ArrayList::new
, List::add
e List::addAll
seja um caso bastante frequente e corriqueiro. Será que não dá para encapsular esses três num objeto só? Bem, é para isso que serve a interface Collector
. E o método collect(Collector)
é a versão de collect
que utiliza tal interface. Para obter então um Collector
que cria a lista usando esses três parâmetros, basta usar o método Collectors.toList()
. Logo, você pode usar o collect
simplesmente assim:
List<String> retornoProcessamento = funcoes
.parallelStream()
.map(f -> f.apply(objetoSobProcessamento))
.collect(Collectors.toList());
Esse último deve ser o código que você queria. Observe que é muito mais simples e direto do que o que você postou em sua pergunta!
Coloquei o código completo e compilável no GitHub.
Por fim, quanto à resposta do OnoSendai que você linkou na pergunta, aquilo é sobre JavaScript, e não Java. O modelo do JavaScript é totalmente single-threaded, então não há muito o que se falar de paralelismo lá.