Agrupar, Achatar, Reduzir — Collectors e Gatherers

As operações que transformam uma stream em estruturas ricas, como mapas agrupados, estatísticas e partições; o achatamento de estruturas aninhadas; a redução a um valor; o processamento em janelas e o paralelismo.

"Pare de Escrever Laços" terminou com o relatório de notas pela metade: a stream lia e validava as notas, mas a média por aluno, a contagem por disciplina e a separação entre aprovados e reprovados ainda exigiriam mapas de listas e laços. Esta aula completa o repertório das streams. Os coletores reúnem os elementos em estruturas prontas, o flatMap achata coleções dentro de coleções, o reduce combina todos os elementos num valor, os gatherers, a novidade mais recente da API, processam os elementos em janelas e sequências, e as streams paralelas dividem o trabalho entre os núcleos do processador.

collect e os coletores

A operação terminal collect recebe um coletor, um objeto que descreve como reunir os elementos. A classe Collectors traz dezenas deles prontos. Os mais simples produzem coleções:

.collect(Collectors.toList())                        // uma List (mutável)
.collect(Collectors.toSet())                         // um Set
.collect(Collectors.toCollection(TreeSet::new))      // a coleção que você escolher
.collect(Collectors.joining(", "))                   // um String

Na prática, toList() diretamente na stream substitui o primeiro, com a vantagem de produzir uma lista imutável. Os coletores ficam mais interessantes a partir daqui.

groupingBy: o SQL das streams

Com as notas válidas do relatório numa lista notas, a média por aluno:

Map<String, Double> mediaPorAluno = notas.stream()
    .collect(Collectors.groupingBy(Nota::aluno, TreeMap::new, Collectors.averagingDouble(Nota::valor)));
IO.println(mediaPorAluno);
// {Ana=8.5, Bruno=6.75, Carla=8.5}

groupingBy recebe até três argumentos:

1. A função de classificação, que diz a qual grupo cada elemento pertence: aqui, o aluno. Ela vira a chave do mapa.
2. A fábrica do mapa, opcional: TreeMap::new para ter as chaves ordenadas. Sem ela, o resultado é um HashMap, sem ordem garantida.
3. O coletor de cada grupo (downstream), opcional, que decide o que fazer com os elementos de cada grupo: aqui, a média das notas. Sem ele, cada grupo vira uma lista.

Esse terceiro argumento é o que torna o groupingBy tão poderoso, porque qualquer coletor pode ocupar esse lugar:

// quantas notas por disciplina
Map<String, Long> porDisciplina = notas.stream()
    .collect(Collectors.groupingBy(Nota::disciplina, Collectors.counting()));
// {Matemática=3, História=2, Física=1}: sem TreeMap::new, vale a ordem do HashMap

// quais alunos em cada disciplina
Map<String, List<String>> alunosPorDisciplina = notas.stream()
    .collect(Collectors.groupingBy(Nota::disciplina, TreeMap::new,
             Collectors.mapping(Nota::aluno, Collectors.toList())));
// {Física=[Bruno], História=[Ana, Carla], Matemática=[Ana, Bruno, Carla]}

counting conta os elementos do grupo. mapping transforma cada elemento antes de passá-lo a outro coletor, aqui a lista, para guardar só os nomes em vez das notas inteiras. Quem conhece SQL reconhece o GROUP BY com COUNT e AVG. A semelhança não é coincidência, e vai ajudar muito na trilha de bancos de dados.

Compare com o mapa de listas de "Recursos que se Fecham Sozinhos", construído com get, teste de null, put e add. Aqui, a mesma estrutura é uma linha que declara o agrupamento desejado.

partitioningBy: dividir em dois

Quando o critério é um teste de sim ou não, partitioningBy divide os elementos em exatamente dois grupos, com as chaves true e false:

Map<Boolean, List<String>> aprovacao = notas.stream()
    .collect(Collectors.partitioningBy(n -> n.valor() >= 7.0,
             Collectors.mapping(n -> n.aluno() + "/" + n.disciplina(), Collectors.toList())));

IO.println("aprovadas: " + aprovacao.get(true));
IO.println("reprovadas: " + aprovacao.get(false));

aprovadas: [Ana/Matemática, Ana/História, Carla/Matemática, Carla/História, Bruno/Física]
reprovadas: [Bruno/Matemática]

A diferença para um groupingBy com um teste booleano: a partição sempre tem as duas chaves, mesmo que um dos grupos fique vazio.

Estatísticas de uma vez

Para várias estatísticas ao mesmo tempo, summarizingDouble percorre os dados uma só vez:

DoubleSummaryStatistics est = notas.stream().collect(Collectors.summarizingDouble(Nota::valor));
IO.println("mín %.1f, máx %.1f, média %.2f, n=%d".formatted(
    est.getMin(), est.getMax(), est.getAverage(), est.getCount()));
// mín 6.5, máx 9.5, média 7.92, n=6

E teeing envia cada elemento para dois coletores e combina os dois resultados no fim:

String extremos = notas.stream().collect(Collectors.teeing(
    Collectors.minBy(Comparator.comparingDouble(Nota::valor)),
    Collectors.maxBy(Comparator.comparingDouble(Nota::valor)),
    (min, max) -> min.get().aluno() + " (" + min.get().valor() + ") a "
                + max.get().aluno() + " (" + max.get().valor() + ")"));
// Bruno (6.5) a Carla (9.5)

toMap e a chave duplicada

toMap constrói um mapa a partir de uma função para a chave e outra para o valor:

record Produto(String codigo, String nome, double preco) { }

List<Produto> produtos = List.of(
    new Produto("A1", "Caneta", 3.5),
    new Produto("B2", "Caderno", 22.0),
    new Produto("A1", "Caneta azul", 3.9));

Map<String, Produto> porCodigo = produtos.stream()
    .collect(Collectors.toMap(Produto::codigo, p -> p));

IllegalStateException: Duplicate key A1 (attempted merging values Produto[codigo=A1, nome=Caneta, ...] and Produto[codigo=A1, nome=Caneta azul, ...])

Um mapa não pode ter duas vezes a mesma chave, e o toMap prefere falhar a escolher sozinho qual valor manter. Se duplicatas são possíveis, um terceiro argumento, a função de combinação, decide o que fazer:

Map<String, Produto> ultimo = produtos.stream()
    .collect(Collectors.toMap(Produto::codigo, Function.identity(), (antigo, novo) -> novo, TreeMap::new));
// {A1=Produto[codigo=A1, nome=Caneta azul, ...], B2=Produto[codigo=B2, nome=Caderno, ...]}

Function.identity() é a função que devolve o próprio elemento, equivalente a p -> p. A combinação (antigo, novo) -> novo mantém o último, e o quarto argumento escolhe o tipo de mapa.

flatMap: achatando estruturas

Às vezes cada elemento contém uma coleção, e o que interessa são os elementos de dentro. Pedidos de uma lanchonete, cada um com vários itens:

record Pedido(String cliente, List<String> itens) { }

List<Pedido> pedidos = List.of(
    new Pedido("Ana", List.of("café", "pão")),
    new Pedido("Bia", List.of("café", "leite", "bolo")),
    new Pedido("Caio", List.of()));

IO.println(pedidos.stream().map(Pedido::itens).toList());
// [[café, pão], [café, leite, bolo], []]

IO.println(pedidos.stream().flatMap(p -> p.itens().stream()).toList());
// [café, pão, café, leite, bolo]

map transforma cada pedido na sua lista de itens, e o resultado é uma lista de listas. flatMap transforma cada pedido numa stream de itens e achata todas essas streams numa só. A partir daí, qualquer operação trabalha com os itens:

Map<String, Long> maisPedidos = pedidos.stream()
    .flatMap(p -> p.itens().stream())
    .collect(Collectors.groupingBy(Function.identity(), TreeMap::new, Collectors.counting()));
// {bolo=1, café=2, leite=1, pão=1}

O padrão aparece o tempo todo: as palavras de várias linhas de texto, os alunos de várias turmas, os arquivos de várias pastas.

List<String> linhas = List.of("o rato roeu", "a roupa do rei", "de roma");
linhas.stream()
    .flatMap(l -> Arrays.stream(l.split(" ")))
    .filter(p -> p.startsWith("r"))
    .toList();
// [rato, roeu, roupa, rei, roma]

reduce: combinando tudo num valor

sum, count e max são casos particulares de uma operação mais geral, a redução: combinar todos os elementos, dois a dois, até sobrar um. reduce recebe um valor inicial, a identidade, e uma função que combina o resultado parcial com o próximo elemento:

List<BigDecimal> precos = List.of(new BigDecimal("19.90"), new BigDecimal("5.10"), new BigDecimal("0.30"));
BigDecimal total = precos.stream().reduce(BigDecimal.ZERO, BigDecimal::add);
// 25.30

É a forma de somar valores BigDecimal, que não têm uma stream especializada como DoubleStream, cumprindo a pendência do exercício 1 da aula anterior. A identidade precisa ser neutra para a operação: zero para soma, um para multiplicação, texto vazio para concatenação. Sem identidade, reduce devolve um Optional, porque uma stream vazia não tem resultado:

Optional<String> maisLonga = Stream.of("sol", "estrela", "lua")
    .reduce((a, b) -> a.length() >= b.length() ? a : b);
// Optional[estrela]

A função de combinação deve ser associativa, ou seja, (a + b) + c tem de dar o mesmo que a + (b + c). Soma, multiplicação, máximo e concatenação são associativas, e subtração não é. A exigência vem do paralelismo: com várias threads, a stream é dividida em pedaços reduzidos separadamente, e os resultados parciais são combinados depois, em uma ordem que não é garantida.

Gatherers: janelas e sequências

Algumas operações não se encaixam bem em map e filter, porque precisam lembrar de elementos anteriores: agrupar de três em três, calcular uma média móvel, acumular um total parcial. Até o Java 23, isso exigia sair da stream. O Java 24 finalizou os gatherers, um mecanismo para criar operações intermediárias novas, e trouxe alguns prontos na classe Gatherers. Eles são aplicados com a operação gather:

List<Integer> vendas = List.of(10, 12, 9, 15, 20, 18, 25);

vendas.stream().gather(Gatherers.windowFixed(3)).toList();
// [[10, 12, 9], [15, 20, 18], [25]]

vendas.stream().gather(Gatherers.windowSliding(3)).toList();
// [[10, 12, 9], [12, 9, 15], [9, 15, 20], [15, 20, 18], [20, 18, 25]]

vendas.stream().gather(Gatherers.scan(() -> 0, Integer::sum)).toList();
// [10, 22, 31, 46, 66, 84, 109]

• windowFixed(n) agrupa os elementos em blocos de n, e o último bloco pode ser menor. É útil para processar dados em lotes, como gravar de 100 em 100 registros num banco.
• windowSliding(n) produz janelas que deslizam uma posição por vez.
• scan produz o acumulado passo a passo, os totais parciais, enquanto o reduce daria só o total final.
• fold faz a redução completa, produzindo um único elemento, mas como uma etapa intermediária.

Com a janela deslizante, a média móvel das vendas, um cálculo clássico de análise de dados, é uma pipeline de três etapas:

List<Double> mediaMovel = vendas.stream()
    .gather(Gatherers.windowSliding(3))
    .map(janela -> janela.stream().mapToInt(Integer::intValue).average().orElse(0))
    .toList();
// [10.33..., 12.0, 14.66..., 17.66..., 21.0]

Também é possível escrever gatherers próprios, com Gatherer.ofSequential, informando um estado inicial e o que fazer com cada elemento. Um que remove repetições consecutivas, diferente do distinct, que remove todas:

static <T> Gatherer<T, ?, T> semRepeticaoSeguida() {
    return Gatherer.ofSequential(
        () -> new Object[] {null},                       // estado: o último elemento visto
        (estado, elemento, saida) -> {
            if (!Objects.equals(estado[0], elemento)) {
                estado[0] = elemento;
                return saida.push(elemento);             // envia adiante
            }
            return true;                                 // descarta e continua
        });
}

Stream.of(1, 1, 2, 2, 2, 3, 1, 1).gather(semRepeticaoSeguida()).toList();
// [1, 2, 3, 1]

Escrever gatherers próprios é raro no dia a dia, mas saber que eles existem evita que você abandone a stream quando uma operação precisa de memória entre os elementos. A API é recente, e bibliotecas e frameworks ainda estão passando a usá-la.

Streams paralelas

Trocar stream() por parallelStream(), ou acrescentar .parallel(), divide a stream entre os núcleos do processador, que processam pedaços ao mesmo tempo. Para trabalho intenso sobre muitos dados, pode reduzir bastante o tempo. Mas o paralelismo traz exigências que a versão sequencial esconde:

List<Integer> naoSeguro = new ArrayList<>();
IntStream.range(0, 10_000).parallel().forEach(naoSeguro::add);
IO.println(naoSeguro.size());

Nas duas execuções feitas para este artigo, o resultado foi 8022 e 9131, e não 10000. Várias threads chamaram add no mesmo ArrayList ao mesmo tempo, e ele não foi feito para isso: elementos se perderam, e às vezes o programa pode até lançar uma exceção. É o efeito colateral que a aula anterior pediu para evitar, agora cobrando o preço. A versão correta deixa a stream produzir a lista, com .boxed().toList(), e aí o paralelismo é seguro, porque cada thread junta os seus resultados e a stream os combina no fim.

Outros cuidados:

• A ordem. forEach numa stream paralela executa em qualquer ordem: IntStream.range(0, 10).parallel().forEach(...) imprimiu 5 6 8 9 7 2 3 4 0 1 numa execução e outra ordem na seguinte. toList() e forEachOrdered preservam a ordem da fonte.
• O custo de dividir. Dividir o trabalho, coordenar as threads e juntar os resultados custa tempo. Para coleções pequenas ou operações leves, a versão paralela é mais lenta.
• As fontes. Arrays, ArrayList e IntStream.range se dividem bem. LinkedList, Stream.iterate e as linhas de um arquivo se dividem mal.
• O pool compartilhado. Todas as streams paralelas do programa usam o mesmo conjunto de threads. Uma operação lenta, como uma chamada de rede dentro de uma stream paralela, pode travar todas as outras.

A regra prática: use streams sequenciais por padrão. Considere a paralela só quando houver muitos dados, operações pesadas de CPU e uma medição mostrando ganho real. Medir desempenho corretamente em Java exige técnicas próprias, porque o JIT e o aquecimento da JVM distorcem medições ingênuas, e é o assunto de "Medir Antes de Otimizar". A concorrência em geral, com todos os seus cuidados, é a Fase 7 inteira.

Uma operação apareceu várias vezes nesta aula e na anterior sem explicação completa: max, minBy, findFirst e reduce sem identidade devolvem um Optional, e o curso vem usando orElse e get sem discutir. Esse tipo existe para resolver um problema muito mais antigo que as streams, o null, que você encontrou em forma de NullPointerException desde "Moldes e Instâncias".

Fontes e leituras recomendadas

• Documentação de Collectors (Java SE 25): https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/stream/Collectors.html. Todos os coletores prontos, com exemplos de composição.
• JEP 485, "Stream Gatherers": https://openjdk.org/jeps/485. A proposta finalizada no Java 24, com a motivação, os gatherers prontos e o modelo para escrever novos.
• Documentação de Gatherers: https://docs.oracle.com/en/java/javase/25/docs/api/java.base/java/util/stream/Gatherers.html. A lista dos gatherers prontos, incluindo o mapConcurrent, que executa transformações em paralelo com virtual threads, assunto de "Um Milhão de Threads".
• Brian Goetz, "Java Streams, Part 4: Parallel stream performance" (IBM Developer): https://developer.ibm.com/articles/j-java-streams-4-brian-goetz/. Quando o paralelismo compensa, com o modelo NQ (número de elementos vezes custo por elemento).
• Richard Warburton, Java 8 Lambdas (O'Reilly, 2014): um livro curto e prático sobre lambdas e streams, com capítulos dedicados aos coletores e ao paralelismo de dados.

Exercícios

Exercício 1

Agrupe as palavras "sol", "lua", "estrela", "mar", "cometa", "céu" pelo tamanho, contando quantas palavras há de cada tamanho, com os tamanhos em ordem crescente.

Map<Integer, Long> porTamanho = Stream.of("sol", "lua", "estrela", "mar", "cometa", "céu")
    .collect(Collectors.groupingBy(String::length, TreeMap::new, Collectors.counting()));
// {3=4, 6=1, 7=1}

Exercício 2

Uma lista de funcionários tem nomes repetidos (homônimos). Explique o que acontece com funcionarios.stream().collect(Collectors.toMap(Funcionario::nome, Funcionario::salario)) e proponha duas soluções com significados diferentes.

Exercício 3

Com a lista de pedidos da aula, calcule o número total de itens pedidos e a lista de clientes que pediram café.

long totalItens = pedidos.stream()
    .mapToLong(p -> p.itens().size())
    .sum();                                              // 5

List<String> pediramCafe = pedidos.stream()
    .filter(p -> p.itens().contains("café"))
    .map(Pedido::cliente)
    .toList();                                           // [Ana, Bia]

Exercício 4

Com as notas do relatório, produza um mapa com o aluno de maior nota em cada disciplina, na forma {Física=Bruno, História=Carla, Matemática=Ana}.

Map<String, String> melhorPorDisciplina = notas.stream()
    .collect(Collectors.groupingBy(Nota::disciplina, TreeMap::new,
        Collectors.collectingAndThen(
            Collectors.maxBy(Comparator.comparingDouble(Nota::valor)),
            maior -> maior.get().aluno())));

Exercício 5

Explique por que IntStream.range(0, 10_000).parallel().forEach(lista::add) com um ArrayList perde elementos, e reescreva o trecho para que funcione corretamente em paralelo.

List<Integer> lista = IntStream.range(0, 10_000).parallel().boxed().toList();