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 só pode ser processado depois de todo o fluxo Stream ter sido 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, os seus valores são explorados, sendo extraídos (pull) um a um do Stream. No caso do observável, é diferente. Assim que este 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 dispõe de 39 métodos. Vamos apresentar alguns deles. Consideremos o código seguinte:
 |
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();
// visualização 1
personnes.stream().forEach(p -> {
System.out.println(p);
});
System.out.println("----------------");
// visualização 2
personnes.stream().forEach(System.out::println);
}
}
- linha 11: instanciamos uma lista de pessoas;
- linha 13: a partir desta lista, cria-se um Stream. Todas as coleções podem ser assim transformadas em fluxos Stream. Isto permite tirar partido de todos os métodos desta classe, que permitem processar os elementos da coleção de forma mais concisa do que com loops. Permite também tirar partido do paralelismo no processamento dos elementos, quando tal for possível;
- linha 13: o método [Stream.forEach] tem a seguinte assinatura:
 |
Vê-se 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()] produz 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 apresenta essa pessoa;
O código anterior pode ser simplificado da seguinte forma (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 este método tem de ter a assinatura correta, neste caso a assinatura do método [Consumer.accept]: void accept(T t). Tal como referido anteriormente, o parâmetro do método [accept] será um tipo [Personne];
Obtenemos os seguintes resultados:
Quando um fluxo Stream é utilizado, deixa de estar disponível. É necessário reconstruí-lo se se pretender voltar a utilizá-lo. Isto é demonstrado pelo código seguinte [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, decide-se construir o Stream uma única 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 pretende utilizar um Stream, é necessário construí-lo, mesmo que já tenha sido construído anteriormente.
5.2. Exemplo-02 - processamento em paralelo dos elementos de um Stream
 |
Consideremos o seguinte código:
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("-----------------");
// visualizaçã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] escreve na consola a cadeia jSON de uma pessoa, bem como o nome do thread de execução no qual a exibição é feita;
- linha 13: apresenta uma lista de pessoas. Note-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 o processamento pode ser feito em paralelo. Aqui, deve-se partir do princípio de que a ordem de exibição não tem importância, uma vez que, num processamento paralelo, não há garantia quanto à ordem de execução das threads. Note-se, além disso, uma sintaxe que se tornará omnipresente tanto para os Stream como para os Observable:
- (continuação)
- O flux produz 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 alterar-se à medida que o fluxo inicial é submetido a diferentes processamentos.
A execução deste código produz o seguinte resultado:
Verifica-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 implicaria abordar as condições que tornam esse processamento possível. Descobrimos, assim, 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 agora.
5.3. Exemplo-03 - processamento em paralelo dos elementos de um Stream
 |
Consideremos o código seguinte (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, este método fornece um novo valor do acumulador. O valor final deste último é o valor devolvido pelo método [reduce]. O código [3] ilustra este 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 detetados pelo JVM;
- linhas 15-18: cria-se uma lista de 10 milhões de números;
- linha 22: a soma destes números é calculada sequencialmente com um único thread;
Obtêm-se os seguintes resultados:
Agora, substituamos 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 com a ajuda de vários threads. Isto é possível porque a ordem de soma dos números não tem importância. Assim, 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 que se seguem, será frequentemente este o caso. A gestão das threads é, por si só, dispendiosa em termos de tempo. É necessário que a operação realizada por cada núcleo seja suficientemente complexa para que o ganho de desempenho se torne evidente. É isso que ilustra o exemplo seguinte (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: volta-se a utilizar o método [reduce], ao qual se fornece como parâmetro a referência do método das linhas 23-29;
- linha 28: o método [bo] fornece a soma dos seus dois parâmetros;
- linhas 24-27: de forma artificial, faz-se com que o thread espere 1 milésimo de segundo para simular um trabalho intensivo;
Obtêm-se então os seguintes resultados:
Agora, se substituirmos a linha 30 pela seguinte:
long somme = nombres.stream().parallel().reduce(0L, bo);
obtêm-se os seguintes resultados:
É claramente visível o ganho de desempenho proporcionado pela execução em paralelo do cálculo da soma. Para o processamento de 8 números:
- o thread sequencial espera 8 vezes 1 milésimo de segundo, ou seja, 8 ms;
- os 8 threads paralelos esperam, simultaneamente, 1 milésimo de segundo cada um (para simplificar), ou seja, um total de 1 milésimo de segundo para os 8 números;
Podemos, portanto, esperar que a execução paralela seja 8 vezes mais rápida do que a execução sequencial. É mais ou menos o que acontece aqui.
5.4. Exemplo-04 - filtrar um Stream
 |
Consideremos o seguinte código:
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 implementar é o seguinte: boolean test(T t);
- o método [filter] devolve os elementos do Stream que satisfazem o Predicate. Serve, portanto, para filtrar o Stream;
Consideremos o seguinte código:
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: apresentam as pessoas com idade <28;
- linhas 18-20: apresentam 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-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 seguinte código:
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 e cujo único método a implementar é: R apply(T t). Verifica-se que, a partir de um tipo T, a função [apply] produz um tipo R. O método [Stream.map] irá, portanto, produzir um fluxo Stream de tipo R a partir de um fluxo de tipo T (fluxo de tipo T significa aqui, por uma imprecisão linguística que iremos manter, um fluxo de elementos de tipo T).
Analisemos agora o código do exemplo:
- linha 14: de uma pessoa p, retém-se apenas o nome. Obtém-se, assim, um fluxo de String;
- linha 14: de uma pessoa p, retém-se apenas o nome. Obtém-se, assim, 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 seguinte:
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 por 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 ordem crescente de idade
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>>, obtém-se um Stream<T>
affiche("flatMap", Stream.of(1, 2, 3).flatMap(i -> Stream.of(i, i + 10)).collect(Collectors.toList()));
// a partir 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>>, obtém-se um DoubleStream e, em seguida, um array
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 cadeia jSON do segundo parâmetro do método;
- linha 24: apresenta a cadeia 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] devolve o primeiro elemento de um fluxo, caso exista. A sua assinatura é a seguinte:
 |
O resultado é do tipo Optional<T>, um tipo introduzido pelo Java 8:
 |
A classe Optional<T> permite gerir de forma diferente os ponteiros *null. Um método que deva devolver um tipo T, que pode ter o valor *null, pode optar por devolver um tipo Optional<T>**. O método [Optional<T>.isPresent()] permite saber se o método devolveu um valor ou não. O código seguinte [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 Optional
// 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> produzido pelo método [findFirst]. Um código 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 devolver qualquer elemento do fluxo. Durante os testes, verifica-se que uma execução sequencial devolve o primeiro elemento do fluxo, enquanto uma execução paralela pode efetivamente devolver qualquer elemento. Isto é demonstrado pelo código seguinte [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 devolveu o elemento 2 da lista de pessoas. Poderia ter sido outro;
- linha 6: a execução sequencial devolveu o primeiro elemento da lista de pessoas;
A utilização do método [findAny] só parece fazer sentido 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. Tal como indicado na documentação acima, a execução deste método em paralelo traz poucos ganhos de desempenho e pode até causar perdas. Com efeito, para ignorar os primeiros n elementos, os threads são obrigados a coordenar-se, o que anula os ganhos de desempenho decorrentes do paralelismo.
O método [skip] devolve um fluxo Stream<Personne> que é transformado num 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 o implementar por conta própria. Neste caso, a implementação utilizada é a [Collectors.toList()]. A [Collectors] é uma classe que possui vários métodos estáticos que implementam o tipo [Collector<T,A,R>]. É aí que se deve procurar em primeiro lugar quando se pretende transformar um Stream numa coleção padrão de Java:
 |
Iremos utilizar alguns destes métodos mais adiante.
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 reter apenas os primeiros n elementos de um fluxo. 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] devolve o número de elementos de um Stream. A execução em paralelo do método não traz qualquer ganho de desempenho, como mostra o código seguinte (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] tem a seguinte assinatura:
 |
O método [max] devolve 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 implementar tem a seguinte assinatura: int compare (T o1, T o2). Este método deve devolver -1 se o1 < o2, 0 se o1.equals(o2), +1 se o1 > o2. A interface funcional Comparator possui vários métodos estáticos por predefiniçã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 produz 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, obteremos a pessoa com a idade mais elevada. 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 seguinte: (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 boolean 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);
// valor máximo dos 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();
// máximo de 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>. Esta interface é normalmente implementada pelo método [Comparator.naturalOrder()] da linha 49. Mas, neste caso, queremos apresentar o thread de execução (linhas 31-33). Por isso, implementamos nós próprios a interface;
- linha 50: pesquisa do max;
Obtenemos 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 foi, portanto, mais lenta. Se passarmos 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 continua, portanto, a ser mais lenta.
Utiliza-se o método [Stream.min] de forma 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 ordem crescente de idade
affiche("personnes par âge croissant",
personnes.stream().sorted(Comparator.comparingInt(Personne::getAge)).collect(Collectors.toList()));
// as pessoas por 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. Este método permite ordenar um Stream de acordo com a ordem do comparador que lhe é passado como parâmetro. Vimos que a interface [Comparator] oferecia, por predefinição, vários métodos estáticos que implementavam os comparadores mais comuns, nomeadamente para números e 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 fornece a idade da pessoa.
A interface [Comparator] não disponibiliza métodos estáticos para comparar cadeias de caracteres. Na linha 5, criamos nós próprios um lambda que implementa o único método desta interface: int compare(T t1, T t2)
(p1, p2) -> p1.getNom().compareTo(p2.getNom())
Esta função lambda compara os nomes das pessoas. Os resultados obtidos são os seguintes:
A execução em paralelo da ordenação não parece ser possível, como mostra o código seguinte (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]. Esta função lambda ordena os números por ordem decrescente e apresenta 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 ordenação do fluxo de números foi realizada com um único thread. Não houve qualquer paralelismo. Ou será que há algo que me está a escapar?
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. Por conseguinte, realizam uma filtragem. Todos os três devolvem um valor booleano:
- anyMatch devolve true se existir pelo menos um elemento do Stream que satisfaça o filtro;
- noneMatch resulta em true se não existir nenhum elemento do Stream que cumpra o filtro;
- allMatch resulta em true se todos os elementos de Stream corresponderem 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. O seu parâmetro é uma implementação da interface [Collector]. A classe [Collectors] disponibiliza vários métodos estáticos que implementam a interface [Collector]. Até agora, utilizámos o método [Collectors.toList()]. Aqui, utilizamos o método estático [Collectors.groupingBy], que cria um dicionário a partir do Stream. A 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, temos de 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 demonstra o exemplo seguinte (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 por 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 por 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:
 |
Verifica-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:
 |
Permite eliminar duplicados de um fluxo. O método [Stream.of] (linha 2) tem a seguinte assinatura:
 |
Permite criar um Stream a partir de valores explicitamente fornecidos. 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] tem 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 de tipo T do fluxo inicial teria dado origem a um elemento Stream<R>. O método [flatMap], por sua vez, produz um tipo Stream<R>. Este método unifica (flatten) os diferentes fluxos Stream<R> num único fluxo. É isso que mostram os resultados da execução do código anterior:
Existem variantes especializadas de [flatMap]:
// a partir 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] tem a seguinte assinatura:
 |
O método [flatMapToInt] aceita como parâmetro uma função que produz como resultado um tipo IntStream como se segue:
 |
IntStream é um fluxo de int. Este tipo é preferível ao tipo Stream<Integer>, uma vez que o seu processamento evita o boxing/unboxing entre os tipos Integer e int. Esta interface herda vários métodos do tipo Stream<T> e adiciona outros, incluindo o método [sum] acima referido, que soma os elementos do IntStream.
O código seguinte ilustra a utilização do método análogo [flatMapToDouble]:
// de um Stream<DoubleStream>, obtém-se um DoubleStream e, em seguida, um array
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 estes dois exemplos, são os seguintes:
O exemplo seguinte mostra os ganhos de desempenho obtidos ao passar 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, substituamos a linha 22 pela seguinte (Exemplo06i2):
long somme = nombres.stream().mapToLong(n -> n.longValue()).sum();
O método Stream<Integer>.mapToLong permite-nos obter um fluxo do tipo LongStream de elementos do tipo primitivo long, que depois somamos com a função sum. Obtêm-se então 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 adaptados a estes 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 neste objeto pode ser obtido com o método [getAsInt()]. A existência de um valor pode ser verificada através do método [isPresent()]. A linha 2 dos resultados não significa que a classe [OptionalInt] tenha campos denominados [asInt, present]. Por predefinição, a biblioteca jSON utiliza todos os métodos públicos getX e isY do objeto a ser serializado em jSON. E, neste caso, existe efetivamente 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 através do 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:
 |
Verifica-se que o objeto IntSummaryStatistics obtido fornece diversas informações sobre o fluxo de números, tais como o número de valores, a soma, o máximo, o mínimo e a média.