Einen eigenen Collector erstellen

Warum einen eigenen Collector erstellen?

Die Collector-Schnittstelle

Die Factory Collector.of()

import java.util.stream.*;

// Collector that builds a comma-separated String
Collector<String, StringBuilder, String> csvCollector = Collector.of(
    StringBuilder::new,                     // supplier
    (sb, s) -> {
        if (sb.length() > 0) sb.append(',');
        sb.append(s);
    },                                       // accumulator
    (a, b) -> {
        if (a.length() > 0 && b.length() > 0) a.append(',');
        a.append(b); return a;
    },                                       // combiner
    StringBuilder::toString                  // finisher
);

String csv = Stream.of("Alice","Bob","Carol").collect(csvCollector);
System.out.println(csv); // Alice,Bob,Carol

Eigener Collector: TopN

import java.util.*;
import java.util.stream.*;

static <T> Collector<T, ?, List<T>> topN(int n, Comparator<T> comp) {
    return Collector.of(
        () -> new ArrayList<T>(),
        (list, e) -> {
            list.add(e);
            list.sort(comp.reversed());
            if (list.size() > n) list.remove(list.size() - 1);
        },
        (a, b) -> {
            a.addAll(b);
            a.sort(comp.reversed());
            while (a.size() > n) a.remove(a.size() - 1);
            return a;
        },
        Collector.Characteristics.UNORDERED
    );
}

TopN-Collector verwenden

List<Integer> top3 = Stream.of(5,2,9,1,7,3,8)
    .collect(topN(3, Comparator.naturalOrder()));
System.out.println(top3); // [9, 8, 7]

Merkmalskennzeichnungen

Eigener Collector: Laufender Durchschnitt

record RunningAvg(long count, double sum) {
    RunningAvg add(double v) { return new RunningAvg(count+1, sum+v); }
    RunningAvg merge(RunningAvg o) { return new RunningAvg(count+o.count, sum+o.sum); }
    double avg() { return count == 0 ? 0 : sum / count; }
}

Collector<Double, RunningAvg[], Double> avgCollector = Collector.of(
    () -> new RunningAvg[]{new RunningAvg(0, 0)},
    (a, v) -> a[0] = a[0].add(v),
    (a, b) -> { a[0] = a[0].merge(b[0]); return a; },
    a -> a[0].avg()
);

double avg = Stream.of(10.0, 20.0, 30.0).collect(avgCollector);
System.out.println(avg); // 20.0

Collector-Schnittstelle direkt implementieren

import java.util.function.*;
import java.util.stream.*;

class StringJoiningCollector implements Collector<String, List<String>, String> {
    private final String delimiter;
    StringJoiningCollector(String delimiter) { this.delimiter = delimiter; }
    public Supplier<List<String>> supplier() { return ArrayList::new; }
    public BiConsumer<List<String>, String> accumulator() { return List::add; }
    public BinaryOperator<List<String>> combiner() {
        return (a,b) -> { a.addAll(b); return a; };
    }
    public Function<List<String>, String> finisher() {
        return list -> String.join(delimiter, list);
    }
    public Set<Characteristics> characteristics() { return Set.of(); }
}

Eigene und integrierte Collector vergleichen

// Instead of custom, compose:
Map<String, IntSummaryStatistics> stats = orders.stream().collect(
    Collectors.groupingBy(Order::status,
        Collectors.summarizingInt(Order::amount))
);
// This replaces many custom collector implementations

Thread-Sicherheit in eigenen Collectors

Praxisbeispiel: Histogramm-Collector

Collector<Integer, Map<Integer, Long>, Map<Integer, Long>> histogram =
    Collector.of(
        LinkedHashMap::new,
        (map, val) -> map.merge(val / 10 * 10, 1L, Long::sum),
        (a, b) -> { b.forEach((k,v) -> a.merge(k,v,Long::sum)); return a; },
        Collector.Characteristics.IDENTITY_FINISH
    );

Map<Integer,Long> hist = Stream.of(5,15,25,12,8,22,18,3)
    .collect(histogram);
hist.forEach((bucket, cnt) -> System.out.println(bucket+"-"+(bucket+9)+": "+cnt));

Kurze Überprüfung

Rückblick: Eigene Collector