Parallel Collections & Dataflow

This content was generated with the assistance of AI and should be verified against the official documentation before being relied on in production.

The reference material consulted while preparing these pages is listed in the section bibliography.

This page is written against Apache Groovy 6.0 (6.0.0, 2026-09-22) and every example was run with Groovy 6.0.0 on JDK 26 (checked on 2026-10-14). Groovy 6.0 brings the concurrency toolkit that used to need the separate GPars library into the language distribution, in the package groovy.concurrent. In the 6.0 documentation these features are incubating, so their method names and behaviour may still change between releases. Virtual threads, which several of the pools use, need JDK 21 or later; on JDK 17 to 20 the same pools fall back to ordinary threads. This page covers the two data-oriented parts of that toolkit: parallel versions of the collection methods (data parallelism) and dataflow variables (declarative ordering between concurrent computations). Actors, agents and active objects are on Actors, Agents & Active Objects, and async/await is on Async/Await. The sequential methods these parallel ones mirror are on Working with Collections. Java readers will recognise parallel streams on Java Streams & Collectors, and the thread-level background is on Java Concurrency Basics.

The examples keep to the greenhouse scenario: a Reading record from the sensors, processed in bulk.

Parallel collection methods

Groovy adds parallel variants of the usual collection methods directly onto Collection, so they need no import. Each is named after its sequential sibling with a Parallel suffix. Under the hood they use Java parallel streams, and they run on a ForkJoinPool so that you can isolate the work from the JVM-wide common pool (next section). The closures you pass are converted to the java.util.function types by SAM coercion, exactly as for any other Java API that takes a functional interface (see Closures).

import java.time.Instant

record Reading(String sensor, Instant at, double temperature, double humidity, double soil) {}

def base = Instant.parse('2026-06-01T06:00:00Z')
def sensors = ['north', 'south', 'east']
def readings = (0..<9).collect { i ->
    new Reading(sensors[i % 3], base.plusSeconds(i * 600), 18 + i * 1.5, 60 - i, 0.30 + i * 0.01)
}

def celsius = readings.collectParallel { it.temperature }
assert celsius == readings*.temperature          // order of the result is kept
println celsius

def hot = readings.findAllParallel { it.temperature > 25 }
println hot*.temperature

println readings.findParallel { it.sensor == 'south' }.temperature
println readings.anyParallel { it.humidity < 53 }
println readings.everyParallel { it.soil > 0.3 }
println readings.countParallel { it.sensor == 'east' }
[18.0, 19.5, 21.0, 22.5, 24.0, 25.5, 27.0, 28.5, 30.0]
[25.5, 27.0, 28.5, 30.0]
19.5
true
false
3

The result of collectParallel and findAllParallel is in the same order as the input, even though the work was spread over several threads. everyParallel is false because the first soil value is exactly 0.30, which is not greater than it.

Source: Groovy documentation: Parallel collections

The methods fall into five families. { } stands for a closure.

Family Methods Sequential counterpart

Transformation

collectParallel { }, collectManyParallel { } (transform and flatten)

collect, collectMany

Filtering

findAllParallel { }, findParallel { }, findAnyParallel { }, grepParallel(filter), splitParallel { }

findAll, find, grep, split

Iteration

eachParallel { }, eachWithIndexParallel { item, idx -> }

each, eachWithIndex

Predicates

anyParallel { }, everyParallel { }, countParallel { }

any, every, count

Aggregation

sumParallel { a, b -> }, injectParallel(seed) { a, b -> }, minParallel { a, b -> }, maxParallel { a, b -> }, groupByParallel { }

sum, inject, min, max, groupBy

Two notes on the names. findAnyParallel returns any matching element and gives no ordering guarantee, so it can be faster than findParallel, which returns the first match in encounter order. And sumParallel, minParallel and maxParallel take a two-argument closure (a binary operator or a comparator), not a one-argument selector like the sequential sum { }.

record Reading(String zone, double temperature, List<String> alerts = []) {}
def rs = [
  new Reading('north', 22.0, ['frost-watch']),
  new Reading('south', 31.5, ['too-hot', 'vent-open']),
  new Reading('east', 19.0),
  new Reading('south', 30.0, ['too-hot']),
]
println rs.collectManyParallel { it.alerts }
println rs.findAnyParallel { it.temperature > 30 }.zone in ['south']
println rs.grepParallel { it.temperature > 20 }*.zone
println([1, 'a', 2.5, 'b', 3].grepParallel(Number))
def (warm, cool) = rs.splitParallel { it.temperature >= 25 }
println "warm=${warm*.zone} cool=${cool*.zone}"
rs.eachWithIndexParallel { r, i -> assert r == rs[i] }
[frost-watch, too-hot, vent-open, too-hot]
true
[north, south, south]
[1, 2.5, 3]
warm=[south, south] cool=[north, east]

grepParallel uses the same isCase matching as grep, so a class, a regular expression, a range or a closure all work as the filter. The aggregation methods need a little more care:

record Reading(String zone, double temperature) {}
def rs = [new Reading('north', 22.0), new Reading('south', 31.5), new Reading('east', 19.0), new Reading('south', 30.0)]

def temps = rs*.temperature
println temps.sumParallel { a, b -> a + b }
println temps.injectParallel(0.0d) { a, b -> a + b }
println rs.minParallel { a, b -> a.temperature <=> b.temperature }
println rs.maxParallel { a, b -> a.temperature <=> b.temperature }.zone
println rs.groupByParallel { it.zone }.collectEntries { zone, list -> [zone, list.size()] }.sort()

def nums = (1..1000).toList()
println nums.inject(0) { a, b -> a - b }
println nums.injectParallel(0) { a, b -> a - b }      // not associative: undefined result
println nums.injectParallel(0) { a, b -> a + b }
102.5
102.5
Reading[zone=east, temperature=19.0]
south
[east:1, north:1, south:2]
-500500
0
500500

The last three lines show the rule from the documentation: the accumulator of injectParallel must be associative, because the runtime splits the data, reduces the parts independently and combines the partial results in an unspecified grouping. Subtraction is not associative, so the parallel result (0 on this run, and it is allowed to differ between runs) is meaningless, while addition is safe.

Source: Groovy documentation: Parallel collection methods

Pools, ParallelScope and ConcurrentConfig

By default the parallel methods run on ForkJoinPool.commonPool(), which the whole JVM shares with parallel streams and other libraries. For isolation, bind a pool of your own for the duration of a block with ParallelScope.withPool; every *Parallel method called inside the block (and every @Parallel loop, below) then uses that pool, and the pool is shut down when the block ends. Pool is a managed thread pool that implements Executor and AutoCloseable, with four factories:

Factory Backing

Pool.cpu()

A ForkJoinPool sized to the available processors, for CPU-bound work

Pool.fixed(n)

A ForkJoinPool with n workers, for CPU-bound work

Pool.io()

Virtual threads on JDK 21 and later, a sized pool of ordinary threads otherwise, for I/O-bound work

Pool.virtual()

A virtual thread per task (JDK 21 and later)

ParallelScope.withPool(4) { ... } is shorthand for a pool with four workers. Global defaults live in ConcurrentConfig: defaultParallelism is the number of available processors plus one (a convention inherited from GPars), and it can be overridden with the system property groovy.concurrent.poolsize or by assigning to it.

import groovy.concurrent.ConcurrentConfig
import groovy.concurrent.ParallelScope
import groovy.concurrent.Pool
import java.util.concurrent.ConcurrentHashMap

println "processors: " + Runtime.runtime.availableProcessors()
println "defaultParallelism: " + ConcurrentConfig.defaultParallelism

// Which threads run the closure?
def common = new ConcurrentHashMap<String, Boolean>()
(1..200).toList().eachParallel { common[Thread.currentThread().name.replaceAll(/\d+/, 'N')] = true }
println "common pool threads: " + common.keySet().sort()

def inScope = new ConcurrentHashMap<String, Boolean>()
def total = ParallelScope.withPool(4) { scope ->
    (1..200).toList().eachParallel { inScope[Thread.currentThread().name.replaceAll(/\d+/, 'N')] = true }
    (1..1000).toList().injectParallel(0) { a, b -> a + b }
}
println "scoped pool threads: " + inScope.keySet().sort()
println "sum: $total"

def pools = [cpu: Pool.cpu(), fixed: Pool.fixed(3), io: Pool.io(), virtual: Pool.virtual()]
pools.each { name, p -> println "$name -> ${p.class.simpleName}" }
pools.values()*.close()

ConcurrentConfig.defaultParallelism = 2
println "after override: " + ConcurrentConfig.defaultParallelism
processors: 10
defaultParallelism: 11
common pool threads: [ForkJoinPool.commonPool-worker-N]
scoped pool threads: [ForkJoinPool-N-worker-N]
sum: 500500
cpu -> DefaultPool
fixed -> DefaultPool
io -> DefaultPool
virtual -> DefaultPool
after override: 2

The processor count is that of the machine that ran the example. The thread names confirm the point of a scope: outside it the work ran on the common pool, inside it on a private ForkJoinPool. ConcurrentConfig.defaultParallelism is a JVM-wide setting, so change it once at start-up rather than from library code.

Source: Groovy documentation: Pool and ParallelScope

@Parallel for loops

The @Parallel annotation (groovy.transform.Parallel) turns a for-in loop into a parallel one with structured completion: every iteration has finished before the next statement runs.

import groovy.transform.Parallel
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

def hits = new AtomicInteger()
def seen = new ConcurrentHashMap<String, Double>()
@Parallel
for (zone in ['north', 'south', 'east', 'west']) {
    seen[zone] = zone.length() * 1.5d
    hits.incrementAndGet()
}
println "after loop: hits=${hits.get()} seen=${new TreeMap(seen)}"
assert hits.get() == 4
after loop: hits=4 seen=[east:6.0, north:7.5, south:7.5, west:6.0]

The compiler rewrites the loop body into a closure passed to eachParallel, so the loop stops being a loop. Three consequences follow, all from the documentation. break and continue are not supported (they are not valid inside a closure and fail compilation). A return in the body leaves the closure, which skips only the current iteration, not the enclosing method. And the loop variable is renamed internally, which shows up when you step through the code in a debugger; if you need the best debugging experience, call eachParallel or collectParallel directly. Inside a ParallelScope.withPool block the loop uses the bound pool. Any state the body shares must be thread-safe, as the next sections explain.

Source: Groovy documentation: @Parallel for loops

Choosing between parallel collections and async/await

Parallel collections and @Parallel run on a ForkJoinPool with roughly one worker per core. That is ideal when each task keeps a core busy (parsing, arithmetic, aggregation) and a trap when tasks wait, because a sleeping or blocked task still occupies a worker.

Workload Right tool Why

CPU-bound (computation, parsing, math)

@Parallel, the *Parallel methods

Work stealing keeps every core busy

I/O-bound (HTTP calls, database queries, file I/O, delays)

async/await, ideally on virtual threads

Many waits can be pending without tying up an operating-system thread

Mixed

async/await with Pool.io()

Virtual threads cope with both compute and waiting

Composing values over a monadic carrier (Optional, Awaitable, Stream)

DO comprehensions from the groovy-macro-library module

Flattens nested flatMap chains into one notation

The difference is easy to measure. Sixteen tasks that each sleep 100 ms behave very differently on a four-worker pool and with async:

import groovy.concurrent.*

long t0 = System.nanoTime()
ParallelScope.withPool(4) { scope ->
    (1..16).toList().eachParallel { Thread.sleep(100) }
}
long blocking = (System.nanoTime() - t0).intdiv(1_000_000)

t0 = System.nanoTime()
def tasks = (1..16).collect { async { Thread.sleep(100); it } }
await(Awaitable.all(*tasks))
long virtual = (System.nanoTime() - t0).intdiv(1_000_000)
println "raw ms: $blocking vs $virtual"
raw ms: 464 vs 125

Four workers sleeping four at a time need four rounds of 100 ms, while the async tasks overlap almost completely. The exact milliseconds vary from run to run; the ratio is the point. await Awaitable.delay(ms) inside a parallel body has the same problem, because await blocks the calling worker, so keep delay-based work outside parallel collections. The async side is covered on Async/Await and, for Kotlin and Java readers, compare Kotlin Coroutines Basics and Java Virtual Threads.

Source: Groovy documentation: Choosing between parallel collections and async/await

Shared state

A parallel body runs on several threads at once, so a plain ArrayList or HashMap written from inside it is a race. There are two good answers: let the collection method build the result (collectParallel, groupByParallel, injectParallel), or, when you really need side effects, write to a concurrent collection or an atomic.

import java.util.concurrent.CopyOnWriteArrayList
record Reading(String zone, double celsius) {}

def rnd = new Random(42)
def zones = ['north', 'south', 'east']
def readings = (1..30_000).collect { new Reading(zones[it % 3], 15 + rnd.nextDouble() * 20) }

def byZone = readings.groupByParallel { it.zone }
def stats = byZone.collectEntries { zone, rs ->
    [zone, [n: rs.size(), mean: rs*.celsius.sum() / rs.size()]]
}.sort()
stats.each { zone, s -> println "${zone.padRight(6)} n=${s.n} mean=${s.mean.round(2)}" }

def serial = readings.groupBy { it.zone }.collectEntries { z, rs -> [z, rs.size()] }.sort()
assert serial == stats.collectEntries { z, s -> [z, s.n] }

def labels = readings.take(5).collectParallel { "${it.zone}:${Math.round(it.celsius)}" }
println labels
def safe = new CopyOnWriteArrayList()
readings.eachParallel { safe << it.celsius }
println "safe size: ${safe.size()}"
east   n=10000 mean=25.02
north  n=10000 mean=25.05
south  n=10000 mean=25.07
[south:30, east:29, north:21, south:21, east:28]
safe size: 30000

Source: Groovy documentation: Use thread-safe collections for shared state

Dataflow variables

Dataflow programming states the dependencies between concurrent computations and lets the runtime order them. A DataflowVariable is a single-assignment variable: it starts unbound, it can be bound exactly once, and a reader that arrives before the value is there waits for it. It implements Awaitable, so it works with await, with Awaitable.all and with then. In the example below the average depends on two sensor values; the order in which the three async blocks actually run does not matter.

flowchart LR N["north: DataflowVariable"] --> A["average: DataflowVariable"] S["south: DataflowVariable"] --> A A --> R["await(average) in the main thread"]
import groovy.concurrent.DataflowVariable

def north = new DataflowVariable()
def south = new DataflowVariable()
def average = new DataflowVariable()

async { average << (await(north) + await(south)) / 2 }   // waits for both inputs
async { south << 31.5 }                                   // bind order does not matter
async { north << 22.0 }
println "average: ${await(average)}"

try {
    north << 99
} catch (IllegalStateException e) {
    println "second bind: ${e.class.simpleName}"
}
println "still ${await(north)}"

def broken = new DataflowVariable()
broken.bindError(new IOException('sensor offline'))
try {
    await(broken)
} catch (IOException e) {
    println "await rethrew: ${e.message}"
}

def raw = new DataflowVariable()
def fahrenheit = raw.then { it * 9 / 5 + 32 }
async { raw << 25 }
println "fahrenheit: ${await(fahrenheit)}"

def a = new DataflowVariable(); def b = new DataflowVariable(); def c = new DataflowVariable()
async { a << 'north' }; async { b << 'south' }; async { c << 'east' }
println await(a, b, c)
println "isBound: ${a.isBound()}"
average: 26.75
second bind: IllegalStateException
still 22.0
await rethrew: sensor offline
fahrenheit: 77
[north, south, east]
isBound: true

  • << is the same as bind(value). Binding twice throws IllegalStateException, and the first value stays.

  • bindError(throwable) binds the variable to a failure: every await on it rethrows the exception, so an error in a producer reaches all consumers.

  • then { } returns a new variable computed from this one once it is bound, which builds a pipeline without any thread management.

  • await(a, b, c) waits for several awaitables and returns their values in argument order, no matter in which order they were bound.

  • A variable is thread-safe: bind it from one thread, read it from any number of others.

If two variables each wait for the other, neither is ever bound and the readers wait forever, so keep the dependency graph acyclic.

Source: Groovy documentation: DataflowVariable

Dataflows

Dataflows removes the declarations. It creates DataflowVariable instances on demand and exposes them as properties: reading a property waits until it is bound, and assigning to a property binds it. This is the shortest way to write a dataflow program in Groovy.

import groovy.concurrent.Dataflows

def df = new Dataflows()
async { df.report = "${df.zone}: ${df.celsius} C" }
async { df.celsius = 22.5 }
async { df.zone = 'north' }
println df.report

def pending = df.getVariable('later')
println "bound before: ${pending.isBound()}"
async { df.later = 42 }
println "awaited: ${await(pending)}, bound after: ${pending.isBound()}"
north: 22.5 C
bound before: false
awaited: 42, bound after: true

getVariable(name) returns the underlying DataflowVariable without blocking, which is what you need to pass it to Awaitable.all, to check isBound(), or to hand it to Java code. Dataflows relies on Groovy’s dynamic property access, so it exists only on the full Groovy runtime and not in the Java-only groovy-concurrent-java module described on Actors, Agents & Active Objects.

Source: Groovy documentation: Dataflows

Plain threads with Thread.start

Everything above builds on threads that the library manages for you. When you just want a thread, the GDK adds closure-taking Thread.start and Thread.startDaemon (and named variants) to the JDK Thread class, and the JDK’s own Thread.startVirtualThread works with a closure too.

def t = Thread.start { println "in ${Thread.currentThread().name}" }
t.join()
def d = Thread.startDaemon('poller') { println "daemon=${Thread.currentThread().daemon} name=${Thread.currentThread().name}" }
d.join()
def log = Collections.synchronizedList([])
def threads = (1..3).collect { n -> Thread.start("zone-$n") { log << n * 10 } }
threads*.join()
println log.sort()
def v = Thread.startVirtualThread { println "virtual=${Thread.currentThread().virtual}" }
v.join()
in Thread-0
daemon=true name=poller
[10, 20, 30]
virtual=true

Raw threads give no result handling, cancellation or structured completion. Reach for them only for the simplest background job, and prefer async, a pool or an actor otherwise. Process-level concurrency is on External Processes.

Source: Groovy GDK: java.lang.Thread

Where the book is outdated

The Manning book Groovy in Action (publisher page in the references) was written for Groovy 1.0, when concurrency meant Thread, synchronized blocks and the java.util.concurrent classes, with GPars as an add-on library that appeared later. Check these points against the current documentation:

  • groovy.concurrent in Groovy 6.0 ships parallel collections, dataflow variables, actors, agents and async/await in the distribution, with no extra dependency. They are incubating in 6.0.

  • The parallel collection methods are plain Collection extension methods and take a ParallelScope instead of the GPars-style withPool wrappers around a library class.

  • Groovy 6.0 needs JDK 17 or later, and virtual-thread-backed pools need JDK 21 or later.

  • GPars features such as dataflow operators and queues, or software transactional memory, are not described in the groovy.concurrent documentation; the actors page has migration notes.

References