StreamLine

StreamLine is a concurrent Stream API for workloads that benefit from virtual threads or a caller-provided ExecutorService. It keeps the fluent Stream style, but gives you explicit control over worker count and chunk size so you can tune how work is scheduled instead of hoping the default pool guesses correctly.

Motivation

Traditional Java streams are great for in-memory collection work, but parallel() is tied to the shared common pool and offers little control when several concurrent workloads compete for the same runtime. StreamLine exists for the cases where every element does real work such as I/O, waits, or heavier transformations and you want isolated, explicit concurrency settings.

Benefits

• High-Performance Streaming: Uses virtual threads by default and supports custom executors when isolation matters.
• Simple API: Offers a straightforward approach to parallel and asynchronous streaming operations.
• Resource Management: Lets callers own executor lifecycle instead of hiding it in a shared pool.
• Enhanced Scalability: Performs exceptionally well under high-load conditions, scaling effectively across multiple cores.
• Pure Java: No external dependencies for a lightweight integration.
• Functional Design: Embraces modern Java functional paradigms.
• No Reflection: Ensures compatibility with GraalVM native images.

Prerequisites

• Java 21 or later and for using Project Loom

Usage

Basic usage with the default virtual-thread executor:

import berlin.yuna.streamline.model.StreamLine;

public class Example {
    public static void main(final String[] args) {
        StreamLine.of("one", "two", "three")
            .threads(-1) // unlimited workers
            .forEach(System.out::println);
    }
}

Bound concurrency with threads(n):

import berlin.yuna.streamline.model.StreamLine;

public class Example {
    public static void main(final String[] args) {
        final var result = StreamLine.range(0, 100)
            .threads(4) // at most 4 workers
            .map(value -> value * 2)
            .toList();

        System.out.println(result.size());
    }
}

Reduce scheduling overhead with chunks(n):

import berlin.yuna.streamline.model.StreamLine;

public class Example {
    public static void main(final String[] args) {
        final var result = StreamLine.range(0, 1_000)
            .threads(8)
            .chunks(32) // each worker drains up to 32 items before taking the next chunk
            .map(Example::loadRemoteValue)
            .toList();

        System.out.println(result.size());
    }

    private static int loadRemoteValue(final int value) {
        return value;
    }
}

Unlimited threads together with chunking means one worker per chunk:

import berlin.yuna.streamline.model.StreamLine;

public class Example {
    public static void main(final String[] args) {
        StreamLine.range(0, 250)
            .threads(-1)
            .chunks(25) // 10 workers for 250 items
            .unordered()
            .forEach(System.out::println);
    }
}

Index-aware terminal operations:

import berlin.yuna.streamline.model.StreamLine;

public class Example {
    public static void main(final String[] args) {
        StreamLine.of("gamma", "beta", "alpha")
            .sorted()
            .forEachOrdered((index, value) -> System.out.println(index + " -> " + value));
    }
}

Use a custom executor when you want separate scheduling or a hard cap:

import berlin.yuna.streamline.model.StreamLine;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class Example {
    public static void main(final String[] args) throws Exception {
        final ExecutorService executor = Executors.newFixedThreadPool(8);
        try {
            final var result = StreamLine.of(executor, "one", "two", "three")
                .threads(4)
                .chunks(2)
                .map(String::toUpperCase)
                .toList();

            System.out.println(result);
        } finally {
            executor.shutdown();
            executor.awaitTermination(5, TimeUnit.SECONDS);
        }
    }
}

Scheduling Rules

• threads(1) runs sequentially.
• threads(n) with n > 1 caps the number of active workers.
• threads(-1) uses unlimited workers and, when combined with chunks(n), submits one worker per chunk.
• chunks(1) behaves like item-by-item scheduling.
• chunks(n) with n > 1 lets each worker process a batch before claiming more work.
• chunks(-1) enables automatic chunk sizing and is the default.
• Negative values always fall back to the -1 behavior instead of throwing.

Ordered vs Unordered

• ordered(true) keeps encounter order in the result.
• unordered() skips result reordering and is usually the better default when downstream code does not care about stable ordering.

When StreamLine Helps

Workload	Java Stream [A]	Java Parallel Stream [B]	StreamLine [C]	Current Median Result
Small in-memory CPU-only mapping	Usually best	Often slightly worse than sequential	Usually unnecessary	A 0.32 ms B 0.67 ms C 3.92 ms
Blocking I/O per element	Usually slow	Good while the common pool is free	Strong fit	A 469.41 ms B 52.35 ms C 13.99 ms
Many concurrent pipelines (commonPoolParallelism() * 4)	Often steadier than parallel	Can self-contend on the shared pool	Strong fit	A 127.66 ms B 457.71 ms C 20.82 ms
Custom executor isolation	No	No	Yes	Only StreamLine lets you isolate the work
Tunable worker count and batching	No	No	Yes	threads(n) and chunks(n) let you shape the load

Performance Notes

• threads(n) controls concurrency.
• chunks(n) controls batching.
• Small collections with very cheap lambdas are often faster with plain loops or standard sequential streams.
• Scalar terminals like count() and numeric reductions now run without materializing the full terminal result first.
• StreamLine becomes more useful when every element does meaningful work and scheduling overhead is not the dominant cost.
• The main value proposition is not "faster than Java streams in every benchmark". The real win is avoiding the shared ForkJoinPool when several parallel stream workloads, frameworks, and libraries start competing for the same small server.
• Blocking or wait-heavy workloads and many concurrent pipelines are where StreamLine should be evaluated first.
• Benchmark through the real public entrypoint for your workload. Synthetic numbers without the real mapper or consumer are theater in a lab coat.

Benchmark Command

Run the opt-in benchmark report:

mvn -q -Dtest=StreamLineBenchmarkTest -Dstreamline.benchmark=true test

The printed report includes:

• a cheap CPU-only single-stream case where plain Java usually wins
• a cheap scalar terminal case that shows the fused count() path
• a blocking single-stream case where StreamLine should shine
• a core-scaled concurrent common-pool case that reflects the "many pipelines on a small server" problem

Limitations

• StreamLine is not compatible with Java 8
• StreamLine is mainly useful when each item does enough work to justify concurrent scheduling
• The concurrent processing does not extend to operations returning type-specific streams like IntStream, LongStream, DoubleStream, OptionalInt, OptionalLong, OptionalDouble, etc.
• StreamLine has more terminal operations than the usual java stream due its simple design - not sure if this is an advantage or disadvantage ^^