Use Java Stream Gatherers and Virtual Threads for Concurrent Work

Java virtual threads, finalized in Java 21, make it practical to run very large numbers of tasks that spend much of their time blocked on I/O. They are lightweight compared with traditional platform threads and are especially useful for straightforward thread-per-task code.

Java streams solve a different problem. A stream pipeline provides a declarative way to transform data, while a parallel stream normally executes work through the common fork/join pool. Parallel streams are generally best suited to CPU-oriented operations that can be split efficiently across processor cores.

That leaves an interesting question: what if a stream pipeline contains blocking work and you want bounded concurrency backed by virtual threads?

A sequential image-processing stream

Consider an application that walks a directory, finds JPEG files and resizes each image. A sequential implementation can look like this:

package ca.bazlur;

import javax.imageio.ImageIO;
import java.awt.image.BufferedImage;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.stream.Stream;

public class ImageProcessor {

    public void processImages(String directory) {
        try (Stream<Path> paths = Files.walk(Path.of(directory))) {
            paths.filter(Files::isRegularFile)
                .filter(this::isJpeg)
                .forEach(this::resizeAndSaveImage);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private boolean isJpeg(Path path) {
        String name = path.toString().toLowerCase();
        return name.endsWith(".jpeg") || name.endsWith(".jpg");
    }

    private void resizeAndSaveImage(Path path) {
        try {
            BufferedImage image = ImageIO.read(path.toFile());

            if (image == null) {
                return;
            }

            BufferedImage resizedImage = resize(image);
            ImageIO.write(resizedImage, "jpeg", path.toFile());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private BufferedImage resize(BufferedImage originalImage) {
        // Implement the resize operation here.
        return originalImage;
    }
}

The stream itself is sequential, so each image completes before processing moves to the next one. Whether concurrency helps depends on what resizeAndSaveImage() actually does. Image decoding and resizing can consume CPU, while reading and writing files can block on I/O.

Why parallelStream is not the same as virtual-thread concurrency

A developer might make the stream parallel:

paths.parallel()
    .filter(Files::isRegularFile)
    .filter(this::isJpeg)
    .forEach(this::resizeAndSaveImage);

That changes how the stream pipeline is evaluated, but it does not turn the operations into virtual-thread tasks. Parallel streams normally use the common ForkJoinPool, whose parallelism is oriented around available processors. That model can be effective for CPU-bound transformations.

For operations that spend significant time waiting on network, filesystem or other blocking I/O, virtual threads offer a different concurrency model. The number of useful concurrent I/O operations can be much larger than the number of CPU cores, although external resources still impose practical limits.

Stream Gatherers and mapConcurrent

Stream Gatherers extend the Stream API with customizable intermediate operations. They first appeared as a preview feature in JDK 22, were previewed again in JDK 23, and became a permanent Java feature in JDK 24.

The built-in Gatherers.mapConcurrent() operation is particularly relevant to virtual threads. Its conceptual signature is:

public static <T, R> Gatherer<T, ?, R> mapConcurrent(
    int maxConcurrency,
    Function<? super T, ? extends R> mapper
)

mapConcurrent() invokes the supplied mapping function concurrently using virtual threads, up to the requested maxConcurrency. It also preserves the encounter order of the stream.

This is not a way to make a parallel stream use virtual threads. Instead, it is a stream gatherer that introduces bounded concurrent mapping into a stream pipeline.

Use mapConcurrent for the blocking operation

The concurrency should surround the expensive or blocking operation itself. Merely mapping each path to itself concurrently does no useful concurrent work.

For example:

import java.util.stream.Gatherers;

paths.filter(Files::isRegularFile)
    .filter(this::isJpeg)
    .gather(Gatherers.mapConcurrent(
        100,
        this::resizeAndSave
    ))
    .forEach(result ->
        System.out.println("Processed: " + result)
    );

The mapping function must return a result, so the image-processing method can return the processed path:

private Path resizeAndSave(Path path) {
    try {
        BufferedImage image = ImageIO.read(path.toFile());

        if (image == null) {
            throw new IllegalArgumentException(
                "Unsupported image: " + path
            );
        }

        BufferedImage resizedImage = resize(image);
        ImageIO.write(resizedImage, "jpeg", path.toFile());

        return path;
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}

Now the image-processing operation itself runs concurrently. The value 100 is an upper bound, not a universally optimal setting. The appropriate value depends on disk characteristics, memory use, downstream services and the amount of CPU work performed by each task.

Parallel streams vs. mapConcurrent

Do not assume more virtual threads means more performance

Virtual threads are inexpensive, but the resources they access are not unlimited. A filesystem can become saturated, a database can exhaust its connection pool, an HTTP service can rate-limit requests, and image processing can still consume substantial CPU and memory.

That is why maxConcurrency matters. It gives the application a way to place a bound around concurrent work even though the underlying tasks use lightweight virtual threads.

Benchmark the complete workload rather than assuming a larger value is faster. For an image-resizing application, for example, storage throughput and CPU-intensive image transformations may become bottlenecks long before virtual-thread creation does.

Which JDK version do you need?

If you are using JDK 24 or newer, Stream Gatherers are a permanent part of the Java platform and no preview flag is required for this API.

JDK 22 introduced Stream Gatherers as a preview feature, and JDK 23 delivered a second preview. Code targeting those releases must account for the preview status and the API version available in that particular JDK.

Java streams and virtual threads work well together

The important distinction is that parallel streams and virtual-thread concurrency solve related but different problems. Parallel streams provide data parallelism, commonly using fork/join execution. Virtual threads make blocking thread-per-task concurrency much cheaper.

Gatherers.mapConcurrent() provides a useful bridge when a stream pipeline contains operations that benefit from bounded virtual-thread concurrency. Use it around the work that actually blocks, choose the concurrency limit based on the resources being accessed, and benchmark the application under realistic conditions.

A N M Bazlur Rahman is a Java Champion and staff software developer at DNAstack. He is also founder and moderator of the Java User Group in Bangladesh.