Write a custom Collector that keeps the n greatest elements
Implement topN: a reusable Collector that reduces a stream to the n greatest elements according to a supplied Comparator, greatest first. Build it with Collector.of — do not delegate to sorted().limit(n).
Constraints:
- it must produce the same result on a parallel stream, so supply a correct combiner
- hold at most
nelements at a time; do not sort or buffer the whole input ngreater than the number of elements returns them all;n <= 0returns an empty list
static <T> Collector<T, ?, List<T>> topN(int n, Comparator<? super T> cmp) {
// your code here
}Write the implementation.
Collector.of(supplier, accumulator, combiner, finisher) builds one. The supplier makes a fresh mutable container per worker, the accumulator folds one element into it, the combiner merges two containers — that is what makes it parallel-safe — and the finisher turns the container into the result. For topN the container is a min-heap: push each element, drop the smallest once it holds more than n, and sort the survivors in the finisher. Memory stays O(n).
- ✗Writing a combiner that drops one side's partial result, so the parallel answer differs from the sequential one
- ✗Buffering every element and sorting at the finisher, which makes memory O(input) instead of O(n)
- ✗Returning the mutable container itself instead of converting it in the finisher
- →What does the
UNORDEREDcharacteristic promise, and when is it safe to declare it? - →When does
IDENTITY_FINISHapply, and what does it save the pipeline?
Solution
static <T> Collector<T, ?, List<T>> topN(int n, Comparator<? super T> cmp) {
return Collector.of(
() -> new PriorityQueue<T>(cmp), // min-heap: the smallest kept element on top
(heap, item) -> { // accumulator: one element
heap.offer(item);
if (heap.size() > n) heap.poll(); // drop the smallest
},
(a, b) -> { // combiner: merge two partial heaps
for (T item : b) {
a.offer(item);
if (a.size() > n) a.poll();
}
return a;
},
heap -> { // finisher: heap → list
List<T> out = new ArrayList<>(heap);
out.sort(cmp.reversed());
return out;
},
Collector.Characteristics.UNORDERED);
}
// usage
List<Employee> top3 = staff.stream()
.collect(topN(3, Comparator.comparingInt(Employee::salary)));Why this shape. The container is a PriorityQueue under the original cmp — a min-heap, so the smallest of the kept elements sits on top. The accumulator pushes an element and, once the size passes n, pops that top. At most n + 1 elements are live at any moment; the input is never sorted whole.
The combiner is not decoration. On a parallel stream each worker accumulates its own heap, and only the combiner can merge them: it pours one heap's elements into the other under the same n bound. A combiner that just returned a would silently lose half the data — and only under parallelism.
Edges. With n <= 0 the heap.size() > n test fires immediately, so the heap stays empty and the result is an empty list. With n larger than the element count nothing is ever dropped and the whole input comes back in descending order. UNORDERED states that encounter order does not affect the result, letting the runtime skip maintaining it.
