Definition
The two grouping collectors differ only when the stream is parallel.
- groupingByConcurrent(classifier) – a concurrent, unordered collector that groups elements into a
ConcurrentMap. - One shared map – on a parallel stream all threads insert into a single concurrent map directly, with no merge step.
- groupingBy (non-concurrent) – in parallel, each thread builds a partial map and the framework merges them, which costs memory and merge time.
- Unordered trade-off – groupingByConcurrent is marked UNORDERED, so the encounter order of elements within a group is not preserved.
- Result type – it returns a
ConcurrentMap, and the classifier and any downstream must be thread-safe.
Many clerks sort transaction slips into labelled account trays at the same time at the National Bank of India. groupingByConcurrent gives them one shared set of trays that is safe for everyone to drop slips into at once. Ordinary groupingBy would give each clerk their own trays, and then someone must merge all the tray sets at the end, which is the slow part when the pile is huge.
Understanding the Concepts
What is groupingByConcurrent?
Collectors.groupingByConcurrent(classifier) is a concurrent, unordered collector that groups elements into a ConcurrentMap, keyed by the classifier. On a parallel stream it lets multiple threads insert into one shared concurrent map simultaneously, avoiding the per-thread-map-then-merge cost that ordinary groupingBy incurs in parallel.
package com.bank.batch;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
public class GroupConcurrent {
record Txn(String accountId, long amount) {}
public static void main(String[] args) {
List<Txn> txns = List.of(
new Txn("ACC-1", 100), new Txn("ACC-2", 200),
new Txn("ACC-1", 50), new Txn("ACC-2", 75));
ConcurrentMap<String, List<Txn>> byAccount = txns.parallelStream()
.collect(Collectors.groupingByConcurrent(Txn::accountId));
System.out.println(byAccount.keySet());
}
}
[ACC-1, ACC-2]
How does it differ from groupingBy on a parallel stream?
Ordinary groupingBy is a non-concurrent collector. On a parallel stream the framework has each thread build its own partial map, then merges those maps at the end. groupingByConcurrent is a concurrent collector, so all threads insert into a single shared ConcurrentMap, skipping the merge.
| Aspect | groupingBy (parallel) | groupingByConcurrent (parallel) |
|--------|-------------------------|-----------------------------------|
| Per-thread maps | yes, one per thread | no, one shared map |
| Merge step | yes, merges all partials | no merge needed |
| Map type | HashMap | ConcurrentHashMap |
| Order within group | preserved | not preserved (unordered) |
| Best when | sequential or ordered groups | large parallel, order irrelevant |
// non-concurrent: per-thread maps merged at the end
Map<String, List<Txn>> a = txns.parallelStream()
.collect(java.util.stream.Collectors.groupingBy(Txn::accountId));
// concurrent: one shared ConcurrentMap, no merge
ConcurrentMap<String, List<Txn>> b = txns.parallelStream()
.collect(java.util.stream.Collectors.groupingByConcurrent(Txn::accountId));
Why does the merge step matter for a very large stream?
For millions of transactions, building a HashMap per worker thread and then merging them costs extra memory for the partial maps and extra time to combine them. The concurrent collector removes that overhead by writing to one map, so it scales better when the data is large and the stream is truly parallel.
package com.bank.batch;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
public class LargeGrouping {
public static void main(String[] args) {
// 5 million synthetic transactions across 1000 accounts
ConcurrentMap<Long, Long> perAccountCount = LongStream.range(0, 5_000_000L)
.parallel()
.boxed()
.collect(Collectors.groupingByConcurrent(
id -> id % 1000,
Collectors.counting()));
System.out.println("accounts = " + perAccountCount.size());
}
}
accounts = 1000
How do you add a downstream collector?
groupingByConcurrent accepts a downstream collector exactly like groupingBy, so you can count, sum, or collect per group. For a total amount per account use a summing collector.
package com.bank.batch;
import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
public class DownstreamConcurrent {
record Txn(String accountId, long amount) {}
public static void main(String[] args) {
List<Txn> txns = List.of(
new Txn("ACC-1", 100), new Txn("ACC-2", 200),
new Txn("ACC-1", 50), new Txn("ACC-2", 75));
ConcurrentMap<String, Long> totalPerAccount = txns.parallelStream()
.collect(Collectors.groupingByConcurrent(
Txn::accountId,
Collectors.summingLong(Txn::amount)));
System.out.println(totalPerAccount);
}
}
{ACC-1=150, ACC-2=275}
What is the unordered trade-off?
groupingByConcurrent is marked UNORDERED, so within each group the encounter order of elements is not preserved. For sums and counts this is irrelevant, but if the order of transactions inside a group matters, the concurrent collector is the wrong choice.
package com.bank.batch;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentMap;
import java.util.stream.Collectors;
public class OrderTradeoff {
record Txn(String accountId, int seq) {}
public static void main(String[] args) {
List<Txn> txns = List.of(
new Txn("ACC-1", 1), new Txn("ACC-1", 2),
new Txn("ACC-1", 3), new Txn("ACC-1", 4));
// ordered: groupingBy preserves encounter order in the list
Map<String, List<Txn>> ordered = txns.parallelStream()
.collect(Collectors.groupingBy(Txn::accountId));
// unordered: groupingByConcurrent may store them in any order
ConcurrentMap<String, List<Txn>> unordered = txns.parallelStream()
.collect(Collectors.groupingByConcurrent(Txn::accountId));
System.out.println("ordered = " + ordered.get("ACC-1"));
System.out.println("unordered = " + unordered.get("ACC-1"));
}
}
ordered = [Txn[accountId=ACC-1, seq=1], Txn[accountId=ACC-1, seq=2], Txn[accountId=ACC-1, seq=3], Txn[accountId=ACC-1, seq=4]]
unordered = [Txn[accountId=ACC-1, seq=3], Txn[accountId=ACC-1, seq=1], Txn[accountId=ACC-1, seq=4], Txn[accountId=ACC-1, seq=2]]
The unordered list order can vary run to run; only use it when order inside a group does not matter.
What are common mistakes with groupingByConcurrent?
The common mistakes are expecting it to speed up a sequential stream and assigning the result to a plain HashMap type.
// WRONG: sequential stream gains nothing from groupingByConcurrent
// and pays for concurrent map overhead
ConcurrentMap<String, List<Txn>> slow = txns.stream() // not parallel
.collect(Collectors.groupingByConcurrent(Txn::accountId));
// WRONG: the result is a ConcurrentMap, not a HashMap
// HashMap<String, List<Txn>> m = txns.parallelStream()
// .collect(Collectors.groupingByConcurrent(Txn::accountId)); // compile error
// CORRECT: parallel stream, result typed as ConcurrentMap or Map
ConcurrentMap<String, List<Txn>> fast = txns.parallelStream()
.collect(Collectors.groupingByConcurrent(Txn::accountId));
Map<String, List<Txn>> asMap = fast; // ConcurrentMap is a Map
What are the thread-safety requirements?
Because many threads insert into the shared map at once, the classifier and any downstream collector must be thread-safe and free of shared mutable state. The built-in collectors such as counting and summingLong are safe; a custom downstream that mutates external state is not.
// SAFE: built-in downstream collectors are thread-safe
ConcurrentMap<String, Long> counts = txns.parallelStream()
.collect(Collectors.groupingByConcurrent(Txn::accountId, Collectors.counting()));
// UNSAFE: a classifier or downstream that touches shared mutable state races
// int[] calls = {0};
// txns.parallelStream().collect(Collectors.groupingByConcurrent(t -> { calls[0]++; return t.accountId(); }));
When should you use groupingByConcurrent and when not?
Use it when the stream is genuinely parallel, the data set is large enough that merge cost matters, order within groups is irrelevant, and the classifier and downstream are thread-safe. Do not use it for sequential streams, for small data sets where parallel overhead dominates, or when you need the encounter order preserved inside each group. In those cases plain groupingBy is correct.
// use groupingByConcurrent: huge parallel count, order irrelevant
ConcurrentMap<String, Long> perAccount = bigTxns.parallelStream()
.collect(Collectors.groupingByConcurrent(Txn::accountId, Collectors.counting()));
// use groupingBy: sequential, or ordered groups needed
Map<String, List<Txn>> ordered = txns.stream()
.collect(Collectors.groupingBy(Txn::accountId));
What are common interview follow-up questions?
Expect: How is groupingByConcurrent different from groupingBy in parallel? Answer: it fills one shared ConcurrentMap with no per-thread merge, while groupingBy builds and merges partial maps. What is the trade-off? Answer: it is unordered, so encounter order within a group is not preserved. What map type does it return? Answer: a ConcurrentMap, typically ConcurrentHashMap. Does it help a sequential stream? Answer: no, it only pays off when the stream is genuinely parallel and large. What must be thread-safe? Answer: the classifier and any downstream collector, because many threads write to the shared map at once.
Visual Overview
groupingBy IN PARALLEL: PER-THREAD MAPS THEN MERGE
transactions.parallelStream()
│ split
┌──────┼──────┐
▼ ▼ ▼
[map1] [map2] [map3] each thread builds its own HashMap
└──────┼──────┘
▼ MERGE (costs memory + time)
┌──────────────┐
│ final HashMap │
└──────────────┘
groupingByConcurrent IN PARALLEL: ONE SHARED MAP, NO MERGE
transactions.parallelStream()
│ split
┌──────┼──────┐
▼ ▼ ▼
thread1 thread2 thread3
└──────┼──────┘
▼ all insert directly (thread-safe)
┌────────────────────┐
│ shared ConcurrentMap│ no merge step
└────────────────────┘
│
▼ UNORDERED: order within each group not preserved
DECISION
parallel + large + order-irrelevant + thread-safe ──▶ groupingByConcurrent
sequential OR ordered groups needed ──▶ groupingBy
Key Takeaway
groupingByConcurrent is a concurrent, unordered collector that groups elements into a single shared ConcurrentMap. On a sequential stream it behaves much like groupingBy, but the difference shows up in parallel. Ordinary groupingBy is non-concurrent, so a parallel stream has each thread build its own partial map and then merges them, which costs memory and merge time. groupingByConcurrent lets all threads insert into one shared concurrent map directly, so that merge step disappears and parallel grouping is faster.
In our work at the National Bank of India, Rajesh grouped a very large end-of-day transaction stream by account to compute per-account activity, and it had to finish inside the batch window using every core. He used a parallel stream with groupingByConcurrent on the account id and a counting downstream, which filled one shared ConcurrentMap across all threads and avoided the per-thread-map merge bottleneck.
The point worth stressing is the trade-off and when not to use it. groupingByConcurrent is marked unordered, so the encounter order of elements within each group is not preserved, which is perfectly fine for counts and sums but wrong when order matters. It only helps when the stream is genuinely parallel and the classifier and downstream are thread-safe, and it returns a ConcurrentMap rather than a plain HashMap.
For a sequential stream, or when ordered groups are needed, plain groupingBy is the right choice, and for small data the overhead of parallelism and a concurrent map is not worth it. Pankaj's rule on the team is to reach for groupingByConcurrent only when the data is large, the stream is parallel, and nobody downstream depends on order.
Remember one sentence: groupingByConcurrent trades order for speed by filling one shared ConcurrentMap from many threads with no merge, so use it only on large parallel streams where order inside a group does not matter.
---