5. O tipo Stream<T> do Java 8
5.1. Exemplo-01 — a classe Stream
As operações nos fluxos Observable têm muitos pontos em comum com os fluxos Stream. Uma diferença é que um elemento de um fluxo Stream não pode ser processado antes que todo o fluxo Stream seja obtido, enquanto um elemento de um fluxo Observable pode ser processado (observado) assim que for obtido, sem esperar pela obtenção do fluxo Observable na íntegra. Outra diferença é que, uma vez obtido o Stream, seus valores são explorados por meio da extração (pull) um a um do Stream. Para o observável, é diferente. Assim que ele emite um valor, este é enviado (pushed) ao seu assinante.
Várias classes implementam o conceito de Stream. Apresentamos aqui a classe Stream<T>:
A classe Stream possui 39 métodos. Apresentaremos alguns deles. Consideremos o código a seguir:
 |
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01 {
public static void main(String[] args) {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// exibição 1
personnes.stream().forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// exibição 2
personnes.stream().forEach(System.out::println);
}
}
- linha 11: instanciamos uma lista de pessoas;
- linha 13: a partir dessa lista, cria-se um Stream. Todas as coleções podem ser transformadas dessa forma em fluxos Stream. Isso permite aproveitar todos os métodos dessa classe, que possibilitam o processamento dos elementos da coleção de maneira mais concisa do que com loops. Além disso, permite aproveitar o paralelismo no processamento dos elementos, quando isso for possível;
- linha 13: o método [Stream.forEach] tem a seguinte assinatura:
 |
Vemos que o parâmetro do método é a interface funcional [Consumer<T>] apresentada no parágrafo 4.4 — uma interface cujo único método utiliza o tipo T e não retorna nada.
- No código:
personnes.stream().forEach(p -> {
System.out.println(p);
});
- [personnes.stream()] gera um fluxo de elementos do tipo [Personne] que alimenta o método [forEach]. O parâmetro p é do tipo [Personne] e a função lambda fornecida exibe essa pessoa;
O código anterior pode ser simplificado da seguinte maneira (linha 18):
personnes.stream().forEach(System.out::println);
Em vez de passar como parâmetro o valor de uma função lambda, passamos a referência de um método existente, neste caso o método println da classe System.out. É claro que esse método deve ter a assinatura correta, neste caso, a assinatura do método [Consumer.accept]: void accept(T t). Conforme mencionado anteriormente, o parâmetro do método [accept] será do tipo [Personne];
Obtemos os seguintes resultados:
Quando um fluxo Stream é processado, ele não pode mais ser processado. É necessário reconstruí-lo caso se deseje processá-lo novamente. Isso é demonstrado pelo código a seguir, [Exemple01b]:
package dvp.java8.streams;
import java.util.stream.Stream;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple01b {
public static void main(String[] args) {
// fluxo de pessoas
Stream<Personne> personnes = Personnes.get().stream();
// visualização 1
personnes.forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// visualização 2
personnes.forEach(System.out::println);
}
}
- linha 11: para otimizar o código, decidimos construir o Stream apenas uma vez. Os resultados obtidos são, então, os seguintes:
{"nom":"jean","age":20,"poids":70.0,"sexe":"HOMME"}
{"nom":"marie","age":10,"poids":30.0,"sexe":"FEMME"}
{"nom":"camille","age":30,"poids":55.0,"sexe":"FEMME"}
----------------
Exception in thread "main" java.lang.IllegalStateException: stream has already been operated upon or closed
at java.util.stream.AbstractPipeline.sourceStageSpliterator(Unknown Source)
at java.util.stream.ReferencePipeline$Head.forEach(Unknown Source)
at dvp.java8.streams.Exemple02b.main(Exemple02b.java:18)
Sempre que se deseja utilizar um Stream, é necessário gerá-lo, mesmo que já tenha sido gerado anteriormente.
5.2. Exemplo-02 – processamento paralelo dos elementos de um Stream
 |
Consideremos o código a seguir:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple02 {
public static void main(String[] args) {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// visualização 1
personnes.stream().forEach(Exemple02::affiche);
System.out.println("-----------------");
// exibição 2
personnes.stream().parallel().forEach(Exemple02::affiche);
}
public static void affiche(Personne p) {
System.out.printf("Personne %s sur thread %s%n", p, Thread.currentThread().getName());
}
}
- linhas 19-21: o método [affiche] exibe no console a string jSON de uma pessoa, bem como o nome do thread de execução no qual a exibição ocorre;
- linha 13: exibe uma lista de pessoas. Observe-se que o parâmetro do método [forEach] é a referência do método estático anterior;
- linha 16: faz-se o mesmo, mas com o método [parallel], solicita-se que o processamento dos elementos do fluxo seja feito em paralelo em várias threads. Nem todo processamento pode ser feito em paralelo. Aqui, deve-se supor que a ordem de exibição não tem importância, pois, em um processamento paralelo, não há garantia quanto à ordem de execução das threads. Observe-se, além disso, uma sintaxe que se tornará onipresente tanto para os Stream quanto para os Observable:
- (continuação)
- flux gera elementos e1 que alimentam o método m1;
- flux.m1, por sua vez, é um fluxo de elementos e2 que alimentam o método m2;
- flux.m1.m2 é um fluxo de elementos e3 que alimentam o método m3;
O tipo dos elementos e1, e2 e e3 pode mudar ao longo dos processamentos pelos quais o fluxo inicial passa.
A execução deste código produz o seguinte resultado:
Percebe-se que a execução paralela (linhas 5-7) ocorreu em três threads diferentes e não respeitou a ordem dos elementos, que é a das linhas 1-3. Neste documento, não nos deteremos muito no processamento paralelo dos elementos de um Stream, pois isso exigiria abordar as condições que tornam esse processamento possível. Descobrimos, então, que poucos processamentos podem ser realizados em paralelo. Um dos que se presta naturalmente ao paralelismo é a soma dos elementos numéricos de um fluxo, que apresentamos a seguir.
5.3. Exemplo-03 – processamento paralelo dos elementos de um Stream
 |
Consideremos o código a seguir (Exemplo 03a):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple03a {
public static void main(String[] args) {
final long limite = 10_000_000L;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// soma dos números – método sequencial
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- na linha 22, utilizamos o método [reduce], cuja assinatura é a seguinte:
 |  |
- o método [reduce] trabalha com elementos do tipo T;
- o método [reduce] aplica o mesmo tratamento a todos os elementos de um fluxo: o valor inicial de um acumulador é fornecido como primeiro parâmetro. Um método que instancia a interface funcional [BinaryOperator] [2] é fornecido como segundo parâmetro: a partir de cada elemento e do acumulador, esse método fornece um novo valor para o acumulador. O valor final deste é o valor retornado pelo método [reduce]. O código [3] explica esse mecanismo. O método [apply] é o método da interface funcional [BinaryOperator] [2];
Voltemos ao código de exemplo:
- linha 12: exibe-se o número de núcleos detectados pelo JVM;
- linhas 15-18: cria-se uma lista de 10 milhões de números;
- linha 22: a soma desses números é calculada sequencialmente com um único thread;
Obtêm-se os seguintes resultados:
Agora, vamos substituir a linha 22 do código pela seguinte (Exemplo03b):
long somme = nombres.stream().parallel().reduce(0L, (s, i) -> s + i);
Solicita-se que os elementos do Stream sejam processados em paralelo por meio de várias threads. Isso é possível porque a ordem de soma dos números não importa. Portanto, podemos atribuir n1 números a um thread T1, n2 números a um thread T2, ... e, por fim, somar os resultados fornecidos por esses diferentes threads. Obtemos, então, os seguintes resultados:
Portanto, praticamente não há ganhos de desempenho. Nos exemplos a seguir, esse será frequentemente o caso. O gerenciamento das threads, por si só, consome tempo. É necessário que a operação realizada por cada núcleo seja suficientemente complexa para que o ganho de desempenho se manifeste. É o que mostra o exemplo a seguir (Exemplo03c):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.function.BinaryOperator;
public class Exemple03c {
public static void main(String[] args) {
final long limite = 10_000L;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// soma dos números – método sequencial
début = new Date().getTime();
BinaryOperator<Long> bo = (s, i) -> {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
}
return s + i;
};
long somme = nombres.stream().reduce(0L, bo);
System.out.printf("somme séquentielle : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- linha 30: utiliza-se novamente o método [reduce], ao qual se fornece como parâmetro a referência do método das linhas 23 a 29;
- linha 28: o método [bo] fornece a soma de seus dois parâmetros;
- linhas 24 a 27: artificialmente, faz-se o thread esperar 1 milissegundo para simular um trabalho intensivo;
Obteve-se, então, os seguintes resultados:
Agora, se substituirmos a linha 30 pela seguinte:
long somme = nombres.stream().parallel().reduce(0L, bo);
obtemos os seguintes resultados:
Percebe-se claramente o ganho de desempenho proporcionado pela execução paralela do cálculo da soma. Para o processamento de 8 números:
- o thread sequencial espera 8 vezes 1 milissegundo, ou seja, 8 ms;
- os 8 threads paralelos esperam, ao mesmo tempo, 1 milissegundo cada um (para simplificar), ou seja, um total de 1 milissegundo para os 8 números;
Portanto, é de se esperar que a execução paralela seja 8 vezes mais rápida do que a execução sequencial. É mais ou menos o que ocorre aqui.
5.4. Exemplo-04 – filtrar um Stream
 |
Consideremos o código a seguir:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// visualizações
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- linha 14: o método [Stream.filter] tem a seguinte assinatura:
 |
- o método [filter] espera como parâmetro uma instância da interface funcional [Predicate] apresentada no parágrafo 4.2 e cujo único método a ser implementado é o seguinte: boolean test(T t);
- o método [filter] retorna os elementos do Stream que satisfazem o Predicate. Ele serve, portanto, para filtrar o Stream;
Consideremos o código a seguir:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple04 {
public static void main(String[] args) {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// visualizações
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(p -> {
System.out.println(p);
});
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(p -> {
System.out.println(p);
});
System.out.println("age < 28 ----------------------");
personnes.stream().filter(p -> p.getAge() < 28).forEach(System.out::println);
System.out.println("poids < 50 ----------------------");
personnes.stream().filter(p -> p.getPoids() < 50).forEach(System.out::println);
}
}
- linhas 14-16: exibem as pessoas com idade <28;
- linhas 18-20: exibem as pessoas com peso <50;
- linha 22: faz o mesmo que as linhas 14-16, mas de forma mais concisa;
- linha 24: faz o mesmo que as linhas 18 a 20, mas de forma mais concisa;
Os resultados da execução são os seguintes:
5.5. Exemplo-05 — criar um Stream<T2> a partir de um Stream<T1>
 |
Consideremos o código a seguir:
package dvp.java8.streams;
import java.util.List;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple05 {
public static void main(String[] args) {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// visualizações
System.out.println("Personne --> String ----------------------");
personnes.stream().map(p -> p.getNom()).forEach(System.out::println);
System.out.println("Personne --> Integer ----------------------");
personnes.stream().map(p -> p.getAge()).forEach(System.out::println);
}
}
- na linha 13, o método [Stream.map] tem a seguinte assinatura:
 |
O parâmetro do método [Stream.map] é uma instância da interface funcional [Function] apresentada no parágrafo 4.3, cujo único método a ser implementado é: R apply(T t). Observa-se que, a partir de um tipo T, a função [apply] produz um tipo R. O método [Stream.map] produzirá, portanto, um fluxo Stream do tipo R a partir de um fluxo do tipo T (fluxo do tipo T significa aqui, por um uso impróprio da linguagem que manteremos, um fluxo de elementos do tipo T).
Vamos agora analisar o código do exemplo:
- linha 14: de uma pessoa p, retém-se apenas o nome. Obtém-se, portanto, um fluxo de String;
- linha 14: de uma pessoa p, retém-se apenas o nome. Obtém-se, portanto, um fluxo de Integer;
Os resultados obtidos são os seguintes:
5.6. Exemplo-06 – outros métodos da classe Stream<T>
 |
Ilustramos alguns dos 39 métodos da classe Stream com o código a seguir:
package dvp.java8.streams;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.DoubleStream;
import java.util.stream.IntStream;
import java.util.stream.Stream;
public class Exemple06 {
// mapeador jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// todas as pessoas
affiche("all", personnes);
// a primeira pessoa
affiche("findFirst", personnes.stream().findFirst().get());
// qualquer pessoa
affiche("findAny", personnes.stream().findAny().get());
// as pessoas, exceto a primeira
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
// as duas primeiras pessoas
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
// o número de pessoas
affiche("count", personnes.stream().count());
// a pessoa mais velha
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
// a pessoa mais leve
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
// a última pessoa em ordem alfabética dos nomes
affiche("nom max", personnes.stream().max((p1, p2) -> p1.getNom().compareToIgnoreCase(p2.getNom())).get());
// a idade total de todas as pessoas
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
// as pessoas por idade crescente
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// há pessoas com mais de 100 anos?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// todas as pessoas têm, no máximo, 100 anos?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// todas as pessoas têm mais de 8 anos?
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
// As pessoas são agrupadas por sexo
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
// remoção de elementos duplicados de uma lista
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
// A partir de um Stream<Stream<T>>, cria-se um Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
// de um Stream<Stream<Integer>>, cria-se um IntStream, cuja soma é calculada
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
// de um Stream<Stream<Integer>>, cria-se um DoubleStream e, em seguida, uma matriz
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
// máximo de um fluxo de inteiros
affiche("reduce Integer::max", Stream.of(1, 10, 8).reduce(Integer::max).get());
// mínimo de um fluxo de Double
affiche("reduce Integer::min", Stream.of(1.5, 10.4, 8.9).reduce(Double::min).get());
// média de um fluxo de inteiros
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// estatísticas de um fluxo de inteiros
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- linhas 72 e 75: exibem a string jSON do segundo parâmetro do método;
- linha 24: exibe a string jSON de todas as pessoas. Obtém-se o seguinte resultado:
5.6.1. [findFirst]
// a primeira pessoa
affiche("findFirst", personnes.stream().findFirst().get());
O método [findFirst] retorna o primeiro elemento de um fluxo, caso exista. Sua assinatura é a seguinte:
 |
O resultado é do tipo Optional<T>, um tipo introduzido pelo Java 8:
 |
A classe Optional<T> permite gerenciar os ponteiros de maneira diferente. Um método que deva retornar um tipo T, que pode ter o valor *null, pode optar por retornar um tipo Optional<T>**. O método [Optional<T>.isPresent()] permite saber se o método retornou um valor ou não. O código a seguir [Exemple06b] ilustra parte do funcionamento do Optional<T>*:
package dvp.java8.streams;
import java.util.Optional;
import com.fasterxml.jackson.core.JsonProcessingException;
public class Exemple06b {
public static void main(String[] args) throws JsonProcessingException {
// opcional sem valor
Optional<Integer> o1 = m1();
System.out.println(o1.isPresent());
affiche(o1);
// opcional com valor
Optional<Integer> o2 = m2();
System.out.println(o2.isPresent());
affiche(o2);
}
private static void affiche(Optional<Integer> o1) {
try {
// recupera-se o valor do opcional
// lança uma exceção se não houver valor
System.out.println(o1.get());
} catch (Throwable th) {
System.out.printf("%s : %s%n", th.getClass().getName(), th.getMessage());
}
}
public static Optional<Integer> m1() {
// sem valor
return Optional.empty();
}
public static Optional<Integer> m2() {
// um valor
return Optional.of(10);
}
}
Os resultados obtidos são os seguintes:
false
java.util.NoSuchElementException : No value present
true
10
Voltemos ao código de ilustração do método [findFirst]:
// a primeira pessoa
affiche("findFirst", personnes.stream().findFirst().get());
- linha 2: para simplificar o código, utilizamos o método [get] sobre o Optional<Personne> gerado pelo método [findFirst]. Um código mais correto exigiria que chamássemos o método [Optional<Personne>.isPresent()] antes de chamar o método [get];
O resultado obtido é o seguinte:
5.6.2. [findAny]
// qualquer pessoa
affiche("findAny", personnes.stream().findAny().get());
O método [findAny] tem a seguinte assinatura:
 |
O método [findAny] pode retornar qualquer elemento do fluxo. Durante os testes, observou-se que uma execução sequencial retorna o primeiro elemento do fluxo, enquanto uma execução paralela pode, de fato, retornar qualquer elemento. Isso é demonstrado pelo código a seguir, [Exemple06c]:
package dvp.java8.streams;
import java.util.List;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import dvp.data.Personne;
import dvp.data.Personnes;
public class Exemple06c {
// mapeador jSON
static private ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// lista de pessoas
List<Personne> personnes = Personnes.get();
// todas as pessoas
affiche("all", personnes);
// qualquer pessoa
affiche("findAny parallèle", personnes.stream().parallel().findAny().get());
// qualquer pessoa
affiche("findAny séquentiel", personnes.stream().findAny().get());
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- linha 22: findAny executado em paralelo;
- linha 24: findAny executado sequencialmente;
Os resultados obtidos são os seguintes:
- linha 4: a execução paralela retornou o elemento 2 da lista de pessoas. Poderia ter sido outro;
- linha 6: a execução sequencial retornou o primeiro elemento da lista de pessoas;
O uso do método [findAny] parece fazer sentido apenas no processamento paralelo de um fluxo.
5.6.3. [skip]
// as pessoas, exceto a primeira
affiche("skip 1", personnes.stream().skip(1L).collect(Collectors.toList()));
O método [skip] tem a seguinte assinatura:
 |
O método [skip] ignora os primeiros n elementos de um fluxo. Conforme indicado na documentação acima, a execução desse método em paralelo traz poucos ganhos de desempenho e pode até mesmo causar perdas. De fato, para ignorar os primeiros n elementos, as threads são obrigadas a se coordenar, o que anula os ganhos de desempenho decorrentes do paralelismo.
O método [skip] retorna um fluxo Stream<Personne>, que é transformado em um tipo List<Personne> pelo método [collect], cuja assinatura é a seguinte:
 |
O método [collect] aceita como parâmetro uma instância do tipo [Collector], cuja assinatura é complexa. Existem implementações predefinidas do tipo [Collector] que, na maioria das vezes, permitem evitar a necessidade de implementá-lo por conta própria. Aqui, a implementação utilizada é [Collectors.toList()]. [Collectors] é uma classe que possui vários métodos estáticos que implementam o tipo [Collector<T,A,R>]. É por aí que se deve começar a procurar quando se deseja transformar um Stream em uma coleção padrão do Java:
 |
Utilizaremos alguns desses métodos posteriormente.
A execução produz o seguinte resultado:
O primeiro elemento da lista (jean) foi omitido.
5.6.4. [limit]
// as duas primeiras pessoas
affiche("limit 2", personnes.stream().limit(2L).collect(Collectors.toList()));
O método [limit] tem a seguinte assinatura:
 |
O método [limit] permite manter apenas os primeiros n elementos de um fluxo. Ele não é adequado para processamento paralelo.
A execução produz o seguinte resultado:
5.6.5. [count]
// o número de pessoas
affiche("count", personnes.stream().count());
O método [count] tem a seguinte assinatura:
 |
O método [count] retorna o número de elementos de um Stream. A execução em paralelo do método não traz ganho de desempenho, conforme mostra o código a seguir (Exemplo06d1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.stream.Stream;
public class Exemple06d1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// contagem dos números — método sequencial
Stream<Long> sNombres = nombres.stream();
début = new Date().getTime();
long count = sNombres.count();
System.out.printf("comptage séquentiel : compteur=%s, durée (ms)=%s%n", count, new Date().getTime() - début);
}
}
- linhas 11-22: cria-se um Stream com 10 milhões de números;
- linhas 22-24: contagem do Stream;
A execução produz o seguinte resultado:
Se substituirmos a linha 22 do código pela seguinte (Exemplo06d2):
Stream<Long> sNombres = nombres.stream().parallel();
obtêm-se os seguintes resultados:
5.6.6. [max, min]
// a pessoa mais velha
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
O método [max] possui a seguinte assinatura:
 |
O método [max] retorna o valor máximo de um fluxo utilizando o comparador que lhe é passado como parâmetro. Comparator é uma interface funcional cujo único método a ser implementado tem a seguinte assinatura: int compare (T o1, T o2). Esse método deve retornar -1 se o1 < o2, 0 se o1.equals(o2), +1 se o1 > o2. A interface funcional Comparator possui vários métodos estáticos padrão que implementam a interface Comparator para os casos mais comuns. Assim, na instrução:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
utilizamos o método estático [Comparator.comparingInt], cuja assinatura é a seguinte:
 |
O tipo ToIntFunction é uma interface funcional:
 |
O método [applyAsInt] da interface funcional ToIntFunction gera um tipo int a partir de um tipo T. Voltemos ao nosso código:
affiche("age max", personnes.stream().max(Comparator.comparingInt(Personne::getAge)).get());
O parâmetro efetivo do método [Comparator.comparingInt] deve ser, neste caso, um lambda Pessoa --> int. Passamos a referência do método [Personne.getAge], que possui exatamente essa assinatura. No final, teremos a pessoa com a idade mais alta. Obtemos um tipo Optional<Personne>, do qual extraímos o valor com o método [Optional.get]. Obtemos o seguinte resultado:
O cálculo do max em paralelo não traz ganhos de desempenho, como mostra o exemplo a seguir: (Exemplo06e1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Stream;
public class Exemple06e1 {
public static void main(String[] args) {
// data
// final long limite = 100L;
// final booleano verbose = true;
final long limite = 10_000_000L;
final boolean verbose = false;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextLong());
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// máximo de números — método sequencial
Stream<Long> sNombres = nombres.stream();
Comparator<Long> compLong = (l1, l2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// comparação
long v1 = l1.longValue();
long v2 = l2.longValue();
if (v1 < v2) {
return -1;
} else {
if (v1 == v2) {
return 0;
} else {
return +1;
}
}
};
début = new Date().getTime();
// maior comprimento = sNombres.max(Comparator.naturalOrder()).get();
long max = sNombres.max(compLong).get();
System.out.printf("%nmax séquentiel : max=%s, durée (ms)=%s%n", max, new Date().getTime() - début);
}
}
- linha 29: temos um fluxo de limite números aleatórios do tipo Long;
- linhas 30-47: a variável lambda compLong implementa a interface Comparator<Long>. Essa interface é normalmente implementada pelo método [Comparator.naturalOrder()] da linha 49. Mas, neste caso, queremos exibir o thread de execução (linhas 31-33). Por isso, implementamos a interface nós mesmos;
- linha 50: busca pelo max;
Obtemos os seguintes resultados:
 |
Se agora substituirmos a linha 27 pela seguinte (Exemplo06e2):
Stream<Long> sNombres = nombres.stream().parallel();
obtêm-se os seguintes resultados:
 |
A execução paralela, portanto, foi mais lenta. Se aumentarmos para 10 milhões de números com verbose=false, obtemos os seguintes resultados:
para a execução sequencial:
para a execução paralela, que, portanto, continua sendo mais lenta.
Utiliza-se o método [Stream.min] de maneira análoga:
// a pessoa mais leve
affiche("poids min", personnes.stream().min(Comparator.comparingDouble(Personne::getPoids)).get());
5.6.7. [reduce]
// a idade total de todas as pessoas
affiche("âge total (reduce)", personnes.stream().map(p -> p.getAge()).reduce(0, (a1, a2) -> a1 + a2));
O método [reduce] foi apresentado no parágrafo 5.3. A linha 2 acima soma as idades de todas as pessoas. O resultado é o seguinte:
5.6.8. [sorted]
// as pessoas por idade crescente
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// pessoas em ordem alfabética dos nomes
List<Personne> lPersonnes=personnes.stream().sorted((p1, p2) -> p1.getNom().compareTo(p2.getNom())).collect(Collectors.toList());
affiche("personnes par ordre alphabétique des noms", lPersonnes);
O método [sorted] (linhas 3 e 5) tem a seguinte assinatura:
 |
O método [sorted] aceita como parâmetro o tipo [Comparator] descrito no parágrafo 5.6.6 para os métodos min e max. Ele permite classificar um Stream na ordem do comparador que lhe é passado como parâmetro. Vimos que a interface [Comparator] oferecia vários métodos estáticos por padrão que implementavam os comparadores comuns, notadamente de números e de cadeias de caracteres. Aqui, utilizamos o método [Comparator.comparingInt], que aceita como parâmetro um tipo ToIntFunction, que é uma interface funcional do método [applyAsInt] com a seguinte assinatura: int applyAsInt(T t). Aqui, o parâmetro efetivo passado ao método [Comparator.comparingInt] na linha 3 é a referência do método [Personne.age], que retorna a idade da pessoa.
A interface [Comparator] não oferece métodos estáticos para comparar cadeias de caracteres. Na linha 5, criamos nós mesmos um lambda que implementa o único método dessa interface: int compare(T t1, T t2)
(p1, p2) -> p1.getNom().compareTo(p2.getNom())
Essa função lambda compara os nomes das pessoas. Os resultados obtidos são os seguintes:
A execução paralela da classificação não parece ser possível, conforme mostra o código a seguir (Exemplo06f1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Date;
import java.util.List;
import java.util.Random;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06f1 {
// mapeador jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// dados
final long limite = 100L;
final boolean verbose = true;
// limite final longo = 10_000_000L;
// final boolean verbose = false;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// ordenação dos números — método sequencial
Stream<Integer> sNombres = nombres.stream();
début = new Date().getTime();
Comparator<Integer> compInt = (i1, i2) -> {
if (verbose) {
// thread
System.out.printf("[%s]", Thread.currentThread().getName());
}
// comparação
int v1 = i1.intValue();
int v2 = i2.intValue();
if (v1 < v2) {
return +1;
} else {
if (v1 == v2) {
return 0;
} else {
return -1;
}
}
};
if (verbose) {
affiche("nombres", sNombres.sorted(compInt).collect(Collectors.toList()));
}
System.out.printf("tri séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- linhas 30-36: constrói-se um fluxo de números aleatórios limite;
- linha 32: passa-se a função lambda compInt (linhas 38-55) para o método [sorted]. Essa função lambda ordena os números em ordem decrescente e exibe o thread que a executa.
Os resultados obtidos são os seguintes:
 |
Se, no código anterior, substituirmos a linha 36 pela seguinte (Exemplo06f2):
Stream<Integer> sNombres = nombres.stream().parallel();
obtêm-se os seguintes resultados:
 |
Descobrimos que, surpreendentemente, a classificação do fluxo de números foi realizada com um único thread. Não houve nenhum paralelismo. Ou será que há algo que está me escapando?
5.6.9. [anyMatch, noneMatch, allMatch]
// há pessoas com mais de 100 anos?
affiche("des personnes de + de 100 ans (anyMatch)", personnes.stream().anyMatch(p -> p.getAge() > 100));
// todas as pessoas têm, no máximo, 100 anos?
affiche("des personnes de + de 100 ans (noneMatch)", personnes.stream().noneMatch(p -> p.getAge() > 100));
// todas as pessoas têm mais de 8 anos?
affiche("des personnes de + de 8 ans (allMatch)", personnes.stream().allMatch(p -> p.getAge() > 8));
Nas linhas 2, 4 e 6, os métodos [anyMatch, noneMatch, allMatch] têm como parâmetro um tipo Predicate descrito no parágrafo 4.2. Portanto, eles realizam uma filtragem. Todos os três retornam um valor booleano:
- anyMatch retorna true se houver pelo menos um elemento do Stream que atenda ao filtro;
- noneMatch resulta em true se não houver nenhum elemento do Stream que atenda ao filtro;
- allMatch gera true se todos os elementos de Stream atenderem ao filtro;
Os resultados obtidos são os seguintes:
5.6.10. [collect(Collectors.groupingBy)]
// As pessoas são agrupadas por sexo
affiche("personnes regroupées par sexe", personnes.stream().collect(Collectors.groupingBy(p -> p.getSexe())));
O método [collect] foi apresentado no parágrafo 5.6.3. Seu parâmetro é uma implementação da interface [Collector]. A classe [Collectors] oferece vários métodos estáticos que implementam a interface [Collector]. Até agora, utilizamos o método [Collectors.toList()]. Aqui, utilizamos o método estático [Collectors.groupingBy], que cria um dicionário a partir do Stream. Sua assinatura é a seguinte:
 |
O método [groupingBy] cria, a partir de um tipo Stream<T>, um tipo Map<K,List<T>>. A chave K é fornecida pelo parâmetro do método [groupingBy], do tipo Function<T,K>, cujo único método tem a assinatura: K apply(T t). Se quisermos criar um dicionário indexado pelo sexo das pessoas, devemos fornecer uma função que gere o sexo a partir de uma pessoa. Aqui, passamos como parâmetro efetivo do método [groupingBy] a referência do método [Personne.getSexe]. Os resultados obtidos são os seguintes:
Na linha 2, temos a cadeia jSON de um dicionário indexado por duas chaves: HOMME e FEMME.
O cálculo paralelo não traz ganhos de desempenho, como mostra o exemplo a seguir (Exemplo06g1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
public class Exemple06g1 {
// melhor jSON
static ObjectMapper jsonMapper = new ObjectMapper();
public static void main(String[] args) throws JsonProcessingException {
// dados
final long limite = 100L;
final boolean verbose = true;
// limite final longo = 10_000_000L;
// final boolean verbose = false;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Integer> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(new Random().nextInt(1000));
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// agrupamento dos números em centenas — método sequencial
Stream<Integer> sNombres = nombres.stream();
Function<Integer, Integer> groupByCent = n -> {
if (verbose) {
System.out.printf("[%s]", Thread.currentThread().getName());
}
return n / 100;
};
début = new Date().getTime();
// Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(número -> número / 100));
Map<Integer, List<Integer>> lNombres = sNombres.collect(Collectors.groupingBy(groupByCent));
System.out.printf("%nregroupement séquentiel : durée (ms)=%s%n", new Date().getTime() - début);
// resultados
if (verbose) {
affiche("nombres regroupés", lNombres);
}
}
public static <T> void affiche(String message, T value) throws JsonProcessingException {
System.out.println(String.format("%s ----", message));
System.out.println(jsonMapper.writeValueAsString(value));
}
}
- linhas 23-38: construção de um fluxo de números limite;
- linha 47: os números são agrupados em centenas. Utiliza-se a função lambda das linhas 39-44 para poder exibir o thread de execução;
Os resultados da execução são os seguintes:
 |
Se, no código, substituirmos a linha 38 pela seguinte (Exemplo06g2):
Stream<Integer> sNombres = nombres.stream().parallel();
obtêm-se os seguintes resultados:
 |
Percebe-se que a execução paralela do agrupamento prejudicou o desempenho.
5.6.11. [distinct]
// remoção de elementos duplicados de uma lista
affiche("distinct", Stream.of(1, 2, 1).distinct().collect(Collectors.toList()));
O método [distinct] tem a seguinte assinatura:
 |
Ele permite eliminar duplicatas de um fluxo. O método [Stream.of] (linha 2) tem a seguinte assinatura:
 |
Ela permite criar um Stream a partir de valores fornecidos explicitamente. Os resultados da execução são os seguintes:
5.6.12. [flatMap]
// de um Stream<Stream<T>>, obtém-se um Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
O método [flatMap] possui a seguinte assinatura:
  |
O método [flatMap] aceita como parâmetro uma função que:
- aceita como parâmetro um elemento do tipo T do Stream;
- retorna como resultado um fluxo Stream<R>;
Se, em vez do método [flatMap], tivéssemos utilizado o método [map] descrito no parágrafo 5.5, o resultado seria um tipo Stream<Stream<R>>, em que cada elemento do tipo T do fluxo inicial teria dado origem a um elemento Stream<R>. O método [flatMap], por sua vez, gera um tipo Stream<R>. Ele unifica (flatten) os diferentes fluxos Stream<R> em um único fluxo. É isso que mostram os resultados da execução do código anterior:
Existem variantes especializadas de [flatMap]:
// de um Stream<IntStream>, obtém-se um IntStream, cuja soma é calculada
affiche("flatMapToInt", Stream.of(1, 2, 3).flatMapToInt(i -> IntStream.of(i, i + 10)).sum());
O método [flatMapToInt] possui a seguinte assinatura:
 |
O método [flatMapToInt] aceita como parâmetro uma função que retorna um tipo IntStream como resultado, conforme segue:
 |
IntStream é um fluxo derivado de int. Esse tipo é preferível ao tipo Stream<Integer>, pois seu processamento evita o boxing/unboxing entre os tipos Integer e int. Essa interface herda vários métodos do tipo Stream<T> e adiciona outros, incluindo o método [sum] mencionado acima, que soma os elementos do IntStream.
O código a seguir ilustra o uso do método análogo [flatMapToDouble]:
// de um Stream<DoubleStream>, cria-se um DoubleStream e, em seguida, uma matriz
affiche("flatMapToDouble", Stream.of(1, 2, 3).flatMapToDouble(i -> DoubleStream.of(i, i * 1.2)).toArray());
O método [DoubleStream.toArray] permite passar de um tipo DoubleStream para um tipo double[].
Os resultados, para esses dois exemplos, são os seguintes:
O exemplo a seguir mostra os ganhos de desempenho obtidos ao mudar de um tipo Stream<Long> para um tipo LongStream (Exemplo06i1):
package dvp.java8.streams;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
public class Exemple06i1 {
public static void main(String[] args) {
final long limite = 10_000_000L;
// número de processadores
System.out.printf("La JVM a détecté [%s] coeurs sur votre machine%n", Runtime.getRuntime().availableProcessors());
// lista de números
long début = new Date().getTime();
List<Long> nombres = new ArrayList<>();
for (long i = 0; i < limite; i++) {
nombres.add(i);
}
System.out.printf("création de la liste des %s nombres en %s ms%n", limite, new Date().getTime() - début);
// soma dos números — método sequencial
début = new Date().getTime();
long somme = nombres.stream().reduce(0L, (s, i) -> s + i);
System.out.printf("somme séquentielle du Stream<Integer> : somme=%s, durée (ms)=%s%n", somme, new Date().getTime() - début);
}
}
- linha 22: cálculo da soma de um fluxo de números do tipo Long;
Obtêm-se os seguintes resultados:
Agora, vamos substituir a linha 22 pela seguinte (Exemplo06i2):
long somme = nombres.stream().mapToLong(n -> n.longValue()).sum();
O método Stream<Integer>.mapToLong permite obter um fluxo do tipo LongStream de elementos do tipo primitivo long, que é então somado com a função sum. Assim, obtemos os seguintes resultados:
O ganho de desempenho é evidente.
5.6.13. métodos de fluxo de números primitivos
// valor máximo de um fluxo de inteiros
affiche("IntStream max", IntStream.of(1, 10, 8).max());
// mínimo de um fluxo de double
affiche("DoubleStream min", DoubleStream.of(1.5, 10.4, 8.9).min());
// média de um fluxo de inteiros
affiche("IntStream average", IntStream.of(1, 10, 8).average().getAsDouble());
// Estatísticas de um fluxo de inteiros
affiche("IntStream summaryStatistics", IntStream.of(1, 10, 8).summaryStatistics());
Os fluxos de valores primitivos (int, long, double) oferecem métodos adequados a esses tipos. O resultado da execução do código anterior é o seguinte:
- O resultado da linha 2 do código é um tipo OptionalInt, análogo ao tipo Optional<Integer>. O valor armazenado nesse objeto pode ser obtido com o método [getAsInt()]. A presença de um valor pode ser verificada com o método [isPresent()]. A linha 2 dos resultados não significa que a classe [OptionalInt] possua campos chamados [asInt, present]. Por padrão, a biblioteca jSON utiliza todos os métodos públicos getX e isY do objeto a ser serializado em jSON. E, neste caso, há de fato um método [getAsInt] e outro método [isPresent], sem que os campos [asInt, present] existam;
- o resultado da linha 4 do código é um tipo OptionalDouble análogo ao tipo Optional<Double>;
- o resultado da linha 6 do código é um tipo OptionalDouble, cujo valor pode ser obtido com o método [getAsDouble()]. O método [average] calcula a média do fluxo de números;
- o resultado da linha 8 do código é um tipo IntSummaryStatistics definido da seguinte forma:
 |
Percebe-se que o objeto IntSummaryStatistics obtido fornece diversas informações sobre o fluxo de números, tais como número de valores, soma, máximo, mínimo e média.