Einsammeln und portionieren mit Stream Collector

Es gibt ein unfehlbares Rezept, eine Sache gerecht unter zwei Menschen aufzuteilen: Einer von ihnen darf die Portionen bestimmen, und der andere hat die Wahl.

Gustav Stresemann

Am Ende mancher Tage kommt man von der Arbeit und Teile des Quellcodes haben sich so in die Großhirnrinde gefressen, dass man einfach eine schönere Variante ersinnen muss. Schon alleine, um den Kollegen zu zeigen, dass Legacy nicht von Lethargie abgeleitet ist.

Die Aufgabenstellung klingt dieses mal trivial. Eine Methode wird benötigt, um ein großes Set in viele kleine zu zerlegen. Die Legacy Variante verwendet eine Third Party Methode, die nur auf Listen arbeitet. Daher wird aus dem Set eine List, daraus eine List von List und daraus am Ende eine Collection von Set. Da das ursprüngliche Set, als Endprodukt eines Stream Ausdrucks das Licht der Welt erblickt, wäre eine dazu passende Lösung vorteilhaft.

Einsammeln am Ende eines Stream geschieht mit einem Collector und in der Standard Bibliothek gibt es eine wahre Fundgrube davon. Einsammeln aller Elemente aus dem Stream als List, Map oder Set, das Finden des größten oder kleinsten Wertes, das Zählen der Elemente oder das Konkatenieren der Elemente zu einer Zeichenkette.

Im folgenden Beispiel werden vier Streams mit den Werten 0 bis 19 erzeugt und diese dann in Quartetts bzw. Trios gruppiert. Die ersten beiden mit der Methode toLists als Collection<List<Integer>> und die anderen mit toSets als Collection<Set<Integer>>.

System.err.println(IntStream.range(0, 20).mapToObj(x -> x).collect(toLists(4)));
System.err.println(IntStream.range(0, 20).mapToObj(x -> x).collect(toLists(3)));

System.err.println(IntStream.range(0, 20).mapToObj(x -> x).collect(toSets(4)));
System.err.println(IntStream.range(0, 20).mapToObj(x -> x).collect(toSets(3)));

Die Ausgabe der vier Aufrufe zeigt die Gruppierung in Dreier- und Vierergruppen.

[[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15], [16, 17, 18, 19]]
[[0, 1, 2], [3, 4, 5], [6, 7, 8], [9, 10, 11], [12, 13, 14], [15, 16, 17], [18, 19]]
[[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13, 14, 15], [16, 17, 18, 19]]
[[0, 1, 2], [4, 5, 3], [8, 6, 7], [9, 10, 11], [12, 13, 14], [16, 17, 15], [18, 19]]

Die beiden Methoden toLists und toSets nutzen die Convenience Methode Collector.of. Dieser werden Funktionen für die einzelnen Verabeitungsschritte beim Einsammeln übergeben. Für uns interessant sind das Erzeugen der temporären Speicherstruktur, das Hinzufügen eines einzelnen Elementes zur Struktur und die Rückgabe des Ergebnisses.

Bevor wir aber ins Detail gehen, gibt es den Lösungsansatz für den Collector kurz erklärt. Zu jedem Element wir ein Zähler generiert und dieser durch die Anzahl der Elemente pro Gruppe geteilt. Diese Ganzzahldivision erzeugt den Zähler für die entsprechende Gruppe. Bei Dreiergruppen fallen die Elemente 0 bis 2 in die Gruppe 0, die Elemente 3 bis 5 in die Gruppe 1 und zuletzt die Elementen 3*n bis 3*n + 2 in der Gruppe n. Um die Verwaltung einfach zu halten, nutzen wir eine Map mit dem Gruppenzähler als Key und die Gruppe als Value.

Hier jetzt aber die Methoden toLists und toSets mit ihren internen Collector.of Aufruf.

public static <T> Collector<T, ?, Collection<List<T>>> toLists(int size) {
  return Collector.of(() -> new Accu<T, List<T>>(count, ArrayList::new), (a, b) -> a.add(b), (a, b) -> {
    throw new UnsupportedOperationException("parallel not support");
  }, x -> x.get());
}

private static <T> Collector<T, ?, Collection<Set<T>>> toSets(int size) {
  return Collector.of(() -> new Accu<T, Set<T>>(count, HashSet::new), (a, b) -> a.add(b), (a, b) -> {
    throw new UnsupportedOperationException("parallel not support");
  }, x -> x.get());
}

Der erste Parameter der Collector.of Methode ist der Supplier für den Accumulator. Der Accumulator kümmert sich um das Einsammeln der einzelnen Objekte. In unserem Beispiel enthält der Akkumulator, also die HashMap zum Speichern der Gruppen und den Zähler für die Elemente und die Anzahl der Elemente pro Gruppe. Da wir die Gruppen als Set oder List erzeugen wollen, bekommt der Akkumulator, dafür eine passende Methode übergeben.

private static class Accu<T, U extends Collection<T>> {
  private Map<Integer, U> map = new HashMap<>();
  private AtomicInteger counter = new AtomicInteger(0);
  private int size;
  private Function<? super Integer, ? extends U> function;

  public Accu(int size, Function<? super Integer, ? extends U> function) {
    this.size = size;
    this.function = function;
  }

  public void add(T b) {
    map.computeIfAbsent(counter.getAndIncrement() / size, function).add(b);
  }

  public Collection<U> get() {
    return map.values();
  }
 }

Der zweite Parameter ist eine Funktion um ein Element unserem Akkumulator hinzuzufügen (a, b) -> a.add(b). Die dritte Parameter wäre eine Funktion um die Ergebnisse zweier Akkumulatoren im Parallelbetrieb zusammenzuführen. Da dies in unserem Beispiel nicht so ohne weiteres funktionieren kann, verbieten wir es.

(a, b) -> {
    throw new UnsupportedOperationException("parallel not support");
}

Der letzte Parameter liefert das Ergebnis zurück, in unserem Fall ist dies die values Collection der internen HashMap.

Damit sind wir auch schon fast am Ende des Beitrags angekommen. Als Neuerung kann ich jetzt noch verkünden, dass mit diesem Beitrag beginnend, die entsprechenden Sourcen von mir auf GitLab abgelegt werden. Allen die etwas mehr wissen wollen, wünsche ich viel Spaß beim Stöbern im Projekt Stream Collector Utilities.

Nachtrag

Im Zuge der Ankündigung von JFrog, ihr Repository JCenter zu schließen, ist die Bibliothek als Java 11 Binary nach Maven Central migriert.

<dependency>
  <groupId>de.schegge</groupId>
  <artifactId>stream-collector-utilities</artifactId>
  <version>0.1.3</version>
</dependency>