Java 8, ¿cómo implementar un Collector?

La versión 8 de Java ha traído grandes cambios para este lenguaje. Entre ellos, los más destacados son las expresiones lambdas y los streams, que aportan al lenguaje características de programación funcional. Pero con tantos cambios, es fácil perderse algunos detalles como el que veremos en este post.

Según la Real Academia Española, una colección es "un conjunto ordenado de cosas, por lo general de una misma clase y reunidas por su especial interés o valor". Pero, ¿qué es aquello que determina lo que puede llegar a ser de “interés” o “valor” para un individuo?

Desde monedas hasta paquetes de tabaco, pasando por carteles de “no molestar”, el ser humano se ha dedicado durante siglos a coleccionar objetos muy distintos unos de otros.

Java8 nos proporciona la clase Collectors, la cual ofrece diferentes implementaciones para agrupar y almacenar la información procedente de un Stream. Cada una de esas implementaciones procede de la interfaz Collector.

Por lo tanto, la interfaz Collector nos va a permitir establecer las reglas de agrupamiento y recolección de los datos del modo que más se ajuste a nuestras necesidades.

Como todo se ve mejor con un ejemplo, ahí va en mío: supongamos una lista de corredores, de la cual queremos obtener el Podium de una carrera. Esto puede hacerse de manera muy sencilla implementando nuestro propio Collector, que retornará un objeto Podium con los tres corredores que hicieron los menores tiempos en la carrera.

¿Cómo lo implementamos?

Accumulator

Lo primero que haremos será definir nuestro Accumulator, instanciando en su constructor un nuevo objeto de salida. En este, implementaremos los métodos que utilizará posteriormente nuestro Collector personalizado.

• Método acumulador: realizaremos la lógica necesaria para tratar posteriormente los resultados. En este caso, primero añadiremos la penalización al tiempo del corredor que recibe. Después, añadiremos ese corredor al Podium.
• Método combinador: en el método de combinación de los datos de salida debemos implementar la lógica que decidirá cómo será el objeto resultante. Este método recibirá un Accumulator, del que tendremos que sacar el objeto final que contiene (ver finisher) y que utilizaremos para combinar con el objeto final del propio Accumulator.

En el caso de ejemplo, ordenaremos los corredores de ambos Podium por tiempo de finalización, penalización y, en caso de empate, el dorsal. El Podium final contendrá los 3 corredores que menos tiempo hicieron y con menos penalizaciones.

• Método finalizador: el método de finalización devolverá el objeto final, en este caso el Podium.

public class RunnerAccumulator {
 private Runner firstRunner;
 private Runner secondRunner;
 private Runner thirdRunner;
 public void accumulate(Runner runner) {
 runner.addPenalty();
 decidePositions(runner);
 }
 public RunnerAccumulator combine(RunnerAccumulator other) {
 Podium podium = other.finish();
 podium.getFirstRunner().ifPresent(this::decidePositions);
 podium.getSecondRunner().ifPresent(this::decidePositions);
 podium.getThirdRunner().ifPresent(this::decidePositions);
 return this;
 }
 public Podium finish() {
 return new Podium(firstRunner, secondRunner, thirdRunner);
 }
 private void decidePositions(Runner runner) {
 if (isFasterThan(runner, firstRunner)) {
 setFirstRunner(runner);
 } else if (isFasterThan(runner, secondRunner)) {
 setSecondRunner(runner);
 } else if (isFasterThan(runner, thirdRunner)) {
 thirdRunner = runner;
 }
 }
 private void setFirstRunner(Runner runner) {
 thirdRunner = secondRunner;
 secondRunner = firstRunner;
 firstRunner = runner;
 }
 private void setSecondRunner(Runner runner) {
 thirdRunner = secondRunner;
 secondRunner = runner;
 }
 private static boolean isFasterThan(Runner runner, Runner accumulatorRunner) {
 return (accumulatorRunner == null || runner.compareTo(accumulatorRunner) < 0);
 }
}

Interfaz Collector

Deberemos crear nuestra clase RunnerCollector, que será la que implementará la interfaz Collector, al que indicaremos:

1. Como objeto de entrada el tipo de la lista que vamos a tratar (en este caso el objeto Runner).
2. Como tipo de acumulación de la operación de reducción, el Accumulator que codificaremos (en este caso el RunnerAccumulator).
3. Como objeto de salida de la operación de reducción, el objeto resultante de la acumulación (en este caso el Podium).

public class RunnerCollector implements Collector<Runner, RunnerAccumulator, Podium> {
 @Override
 public Supplier<RunnerAccumulator> supplier() {
 return () -> new RunnerAccumulator();
 }
 @Override
 public BiConsumer<RunnerAccumulator, Runner> accumulator() {
 return RunnerAccumulator::accumulate;
 }
 @Override
 public BinaryOperator<RunnerAccumulator> combiner() {
 return RunnerAccumulator::combine;
 }
 @Override
 public Function<RunnerAccumulator, Podium> finisher() {
 return RunnerAccumulator::finish;
 }
 @Override
 public Set<Characteristics> characteristics() {
 Set<Characteristics> chars = new HashSet<Collector.Characteristics>();
 chars.add(Characteristics.CONCURRENT);
 return chars;
 }
}

Una vez hemos indicado los tipos que intervienen en el Collector, lo siguiente es implementar los métodos que define la interfaz:

1. supplier: en el supplier es donde crearemos la instancia del RunnerAccumulator que implementamos anteriormente.
2. accumulator: en el accumulator devolveremos el método accumulate que definimos en el RunnerAccumulator implementado.
3. combiner: en el combiner devolveremos el método combine que definimos en el RunnerAccumulator.
4. finisher: en el finisher, igual que en el accumulator y en el combiner, devolvemos el método finish definido en nuestro RunnerAccumulator.
5. characteristics: en el characteristics indicaremos las características que tendrá nuestro collector. La interfaz provee un enumerado de características. Las disponibles son:
6. CONCURRENT
7. IDENTITY_FINISH
8. UNORDERED

Nosotros hemos puesto la característica CONCURRENT para indicar que se trata de un Collector concurrente, lo que significa que el contenedor de resultados puede admitir que la función del acumulador se llame simultáneamente con el mismo contenedor de resultados de varios subprocesos.

Utilizando nuestro Collector

Nuestro stream de runner lo paralelizaremos (parallelStream), ya que nuestra implementación de Collector permite concurrencia. Al usar la ejecución en paralelo es muy importante implementar el método combine (en el caso de no ser parallel no es invocado).

public static void main(String[] args) {
 RunnerCollector usersCollector = new RunnerCollector();
 Set<Characteristics> characteristics = usersCollector.characteristics();
 Podium podium = getMockRunners().parallel()
 .collect(Collector.of(usersCollector.supplier(),
 usersCollector.accumulator(),
 usersCollector.combiner(),
 usersCollector.finisher(),
 characteristics.toArray(new Characteristics[characteristics.size()])));
 System.out.println(podium.toString());
 }

El resultado de esta ejecución es un objeto Podium con los tres Runners ganadores:

1º Runner [dorsal=1, name=Mario, surname=Sanchez, time=300, penalty=2, endTime=302] 
2º Runner [dorsal=5, name=Juan, surname=Fernandez, time=308, penalty=0, endTime=308] 
3º Runner [dorsal=2, name=Daniel, surname=Jimenez, time=307, penalty=1, endTime=308]

Conclusiones

Como puede verse, la interfaz Collector es muy útil para agrupar los datos de un Stream de una forma muy sencilla adecuándose a nuestras necesidades.