Alright, let's dive into something that usually feels like a black box in the Stream API: Custom Collectors. You've probably used Collectors.toList() or groupingBy() a thousand times, and for 95% of tasks, those are enough. But eventually, you'll hit a wall where you need to aggregate data into a specialized object, and doing it with a .forEach() or a messy .reduce() starts feeling... well, wrong.

The limitation of the defaults

Let's say we're working on a fintech app. We have a Transaction record, and we want to generate a DailySummary that tracks the total volume and the single largest transaction of the day. I could do this with two separate streams, but that's inefficient. I could use a reduce, but reduce expects the result type to be the same as the element type, or it gets clunky with identity objects.

record Transaction(double amount, String category) {}
record DailySummary(double totalVolume, double maxTransaction) {}

List<Transaction> txns = List.of(
    new Transaction(100.0, "Food"),
    new Transaction(250.0, "Tech"),
    new Transaction(50.0, "Food")
);

My first instinct might be to just use a forEach and a mutable object outside the stream. But that's a side-effect, and it'll blow up the moment we switch to .parallelStream(). I want something that fits naturally into the .collect() terminal operation.

Wiring up Collector.of

Now, I could implement the Collector interface manually, but that's a lot of boilerplate. Instead, I'll use Collector.of(). This is the "secret weapon" for custom collectors. It asks for four things: a supplier, an accumulator, a combiner, and a finisher.

I'll start by creating a simple helper class to hold the mutable state during the collection process, because the accumulator needs to modify something.

class SummaryAccumulator {
    double sum = 0;
    double max = 0;

    void add(Transaction t) {
        sum += t.amount();
        max = Math.max(max, t.amount());
    }

    void combine(SummaryAccumulator other) {
        this.sum += other.sum;
        this.max = Math.max(this.max, other.max);
    }
}

Now, let's try to plug this into Collector.of(). I'll try to map this directly to our DailySummary record.

Collector<Transaction, SummaryAccumulator, DailySummary> summaryCollector = 
    Collector.of(
        SummaryAccumulator::new,           // Supplier: Create the container
        SummaryAccumulator::add,           // Accumulator: How to add an element
        SummaryAccumulator::combine,       // Combiner: How to merge two containers
        acc -> new DailySummary(acc.sum, acc.max) // Finisher: Transform to final result
    );

DailySummary result = txns.stream().collect(summaryCollector);
System.out.println(result); // DailySummary[totalVolume=400.0, maxTransaction=250.0]

The Parallelism Trap

At first glance, the combine method feels like overkill. If I'm running a sequential stream, combine is never even called. I'm tempted to just put (a, b) -> a in there to save time. But that's a dangerous habit.

I decided to test this with a parallelStream(). If I had ignored the combiner, my total volume would be wildly incorrect because different threads would be calculating sums in their own buckets, and they'd never merge. By implementing SummaryAccumulator::combine, Java can split the list of transactions across multiple cores and then "fold" those summaries back together perfectly.

One thing to note: the "Finisher" is where the magic happens. If my accumulator was already the final type I wanted (e.g., if I was collecting into a List), I could just use Function.identity(). But since I want an immutable DailySummary record at the end, the finisher is where I convert my mutable "working" object into my clean, immutable "result" object.

When to actually do this

You might be thinking, "Why not just use Collectors.summarizingDouble()?" In this specific case, you probably should. But what if your summary logic involves complex business rules? Maybe the maxTransaction only counts if it's from a specific category, or you're building a complex histogram. That's when you stop looking for a built-in collector and start writing your own.

📋 Practical Task

Exercise: Building a Transactional Volume Aggregator

You need to create a custom collector that aggregates Transaction objects into a CategoryReport. However, there's a catch: the report should only track the total amount and the count of transactions for a specific category provided at runtime.

Requirements:

• Create a Transaction record with double amount and String category.
• Create a CategoryReport record with double totalAmount and long count.
• Implement a static method createCategoryCollector(String targetCategory) that returns a Collector<Transaction, ?, CategoryReport>.
• The collector must ignore any transactions that do not match the targetCategory.
• The collector must be compatible with parallel streams (ensure the combiner is correctly implemented).

Test Case:

List<Transaction> txns = List.of(
    new Transaction(10.0, "Food"),
    new Transaction(20.0, "Food"),
    new Transaction(100.0, "Tech"),
    new Transaction(5.0, "Food")
);

CategoryReport foodReport = txns.parallelStream()
                                .collect(createCategoryCollector("Food"));
// Expected: CategoryReport[totalAmount=35.0, count=3]