Reactive’s Looming Doom. Half I: Evolution – Java Code Geeks

1. Introduction

Multithreading in Java has come an extended evolutionary means from the unique monitor idea and threads mapped to native OS threads to fashionable asynchronous libraries and implementation of light-weight threads (previously “fibers”) as a part of Venture Loom. On the similar time, whereas the language was evolving, the group was creating its personal frameworks and practices. With the discharge of Java 8 and vast adoption of some components of useful programming, frameworks primarily based on rules of reactive methods described within the Reactive Manifesto and, on the API degree, in Reactive Streams Specification, started to achieve traction.

On this collection of posts, we are going to focus on the sensible benefits and downsides of reactive frameworks, what duties they’re greatest fitted to and what future is looming forward for the Reactive strategy.

To achieve a deep understanding of the difficulty, it is very important know the historical past of the selections taken. This publish will reveal the evolution of multithreaded APIs. Within the offered examples the code represents practices of multithreading programming which can be aligned with the API out there on the time, so it received’t use the ConcurrentHashMap#compute() strategies in Java 5 code, and so forth.

2. Problems with the standard strategy to multithreading

Within the conventional strategy, to work with every of those duties, the developer has to make use of an intermediate state for information change between threads. And since it is a shared mutable state in a multithreaded atmosphere, a necessity for a cautious synchronization of entry to that shared state naturally arises.

Let’s see this in observe with an instance of a parallelization drawback. Suppose you’ve got a activity to search out essentially the most frequent identify in an inventory of million names. Single-threaded implementation is sort of trivial: it is sufficient to carry out one iteration over the array and combination the rely of a reputation encountered within the checklist in a Map:

// Generate names
Random r = new Random();
var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
var namesList = IntStream.vary(0, 1000000)
    .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
    .gather(Collectors.toList());

// Mixture counts
Map counts = new HashMap();
for (String identify: namesList)
{
    counts.compute(identify, (n, c) -> c == null ? 1L : c + 1);
}

// Discover the max rely
String mostFrequentName = counts.entrySet()
    .stream()
    .max(Map.Entry.comparingByValue())
    .get()
    .getKey();

System.out.printf("Essentially the most frequent identify is %spercentn", mostFrequentName);

3. Paleozoic

public class Fundamental
{
    public static void primary(String[] args) throws InterruptedException, ExecutionException
    {
        // Generate names
        Random r = new Random();
        var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
        var namesList = IntStream.vary(0, 1000000)
            .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
            .gather(Collectors.toList());

        // Mixture counts
        int batchSize = 1000;
        ArrayList duties = new ArrayList();
        Map finalCounts = new Hashtable();
        for (int i = 0; i < namesList.measurement(); i += batchSize)
        {
            int batchStart = i;
            int batchEnd = Math.min((batchStart + batchSize), namesList.measurement());
            last Record batch = namesList.subList(batchStart, batchEnd);
            // Break up into batches 
            last CountTask activity = new CountTask(batch, finalCounts);
            duties.add(activity);
            activity.setDaemon(true);
            activity.begin();
        }

        // Wait till the threads completed
        for (Thread thread: duties)
        {
            thread.be a part of();
        }

        // Discover the max rely
        String mostFrequentName = finalCounts.entrySet()
            .stream()
            .max(Map.Entry.comparingByValue())
            .get()
            .getKey();

        System.out.printf("Essentially the most frequent identify is %spercentn", mostFrequentName);
    }

    non-public static class CountTask extends Thread
    {
        non-public last Record batch;
        non-public last Map finalCounts;

        non-public CountTask(Record batch, Map finalCounts)
        {
            this.batch = batch;
            this.finalCounts = finalCounts;
        }

        @Override
        public void run()
        {
            Map localCounts = new Hashtable();
            System.out.printf("[%s] Processing batch... n", Thread.currentThread().getName());
            for (String identify: batch)
            {
                localCounts.compute(identify, (n, c) -> c == null ? 1L : c + 1);
            }
            for (Map.Entry stringLongEntry: localCounts.entrySet())
            {
                synchronized (finalCounts)
                {
                    last Lengthy existingCount = finalCounts.get(stringLongEntry.getKey());
                    last var newCount = stringLongEntry.getValue();
                    if (existingCount == null)
                    {
                        finalCounts.put(stringLongEntry.getKey(), newCount);
                    }
                    else
                    {
                        finalCounts.put(stringLongEntry.getKey(), existingCount + newCount);
                    }
                }
            }
        }
    }
}

4. Mesozoic

To cope with the primary challenge, it is sufficient to use thread swimming pools, which have been added to the usual library in Java 5 in the type of the ExecutorService class and a set of its commonplace implementations in the Executors class. To resolve the second one, the Future class was added in the identical launch, that the ExecutorService produces. The Future represents an asynchronous computation with the solely technique to get the result’s to name a blocking methodology Future#get. With all these additions, the Java 5 answer seems to be one thing like this:

public class Fundamental
{
    public static void primary(String[] args) throws InterruptedException, ExecutionException
    {
        // Generate names
        Random r = new Random();
        var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
        var namesList = IntStream.vary(0, 1000000)
            .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
            .toList();

        int batchSize = 1000;
        int parallelism = (int) Math.ceil(namesList.measurement() / (double) batchSize);
        System.out.printf("Parallelism is %s n", parallelism);
        ExecutorService executorService =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(),
                new ThreadFactory()
                {
                    @Override
                    public Thread newThread(Runnable runnable)
                    {
                        Thread t = new Thread(runnable);
                        t.setDaemon(true);
                        return t;
                    }
                });

        // Break up names into batches
        Map finalCounts = new ConcurrentHashMap();
        Record<Callable> duties = new ArrayList();
        for (int i = 0; i < namesList.measurement(); i += batchSize)
        {
            int batchStart = i;
            int batchEnd = Math.min((batchStart + batchSize), namesList.measurement());
            last Record batch = namesList.subList(batchStart, batchEnd);
            duties.add(new CountTask(batch, finalCounts));
        }

        // Wait till duties are finished
        executorService.invokeAll(duties);


        // Discover the max rely
        String mostFrequentName = finalCounts.entrySet()
            .stream()
            .max(Map.Entry.comparingByValue())
            .get()
            .getKey();

        System.out.printf("Essentially the most frequent identify is %spercentn", mostFrequentName);
    }

    non-public static class CountTask implements Callable
    {
        non-public last Record batch;
        non-public last Map finalCounts;

        non-public CountTask(Record batch, Map finalCounts)
        {
            this.batch = batch;
            this.finalCounts = finalCounts;
        }

        @Override
        public Void name()
        {
            Map localCounts = new HashMap();
            System.out.printf("[%s] Processing batch... n", Thread.currentThread().getName());
            for (String identify: batch)
            {
                localCounts.compute(identify, (n, c) -> c == null ? 1L : c + 1);
            }
            for (Map.Entry stringLongEntry: localCounts.entrySet())
            {
                synchronized (finalCounts)
                {
                    last Lengthy existingCount = finalCounts.get(stringLongEntry.getKey());
                    last var newCount = stringLongEntry.getValue();
                    if (existingCount == null)
                    {
                        finalCounts.put(stringLongEntry.getKey(), newCount);
                    }
                    else
                    {
                        finalCounts.put(stringLongEntry.getKey(), existingCount + newCount);
                    }
                }
            }
            return null;
        }
    }
}

This code makes use of Runtime.getRuntime().availableProcessors() threads, thus optimally consuming assets, nevertheless, conceptually it’s virtually similar to the Thread answer. Furthermore, the API will not be optimized particularly for this activity: you must implement a workaround in a type of implementing a Callable<Void> class and producing Future<Void> from the Executor with a purpose to await the computations.

5. Fashionable historical past

Code from the final instance solved virtually each challenge however remained fairly cumbersome and required specific ExecutorService#invokeAll to gather the outcome. So as to deal with that, Java 8 offered a CompletbleFeature class which works particularly nicely with newly launched lambdas. The important thing API distinction from the Future is a risk to offer an asynchronous callback to invoke when the result’s prepared as an alternative of blocking on Future#get.

Additionally, for the CompletbleFeature dependency on the ExecutorService was inverted: as an alternative of manufacturing CompletbleFeatures through the ExecutorService, the Executor was injected into the CompletbleFeature on its creation. It allowed to simply configure which Executor to make use of for a particular CompletbleFeature, and even not present the Executor in any respect: by default CompletbleFeature are executed on ForkJoinPool.

The next code illustrates an answer that employs the CompletbleFeature:

public class Fundamental
{
    public static void primary(String[] args)
    {
        // Generate names
        Random r = new Random();
        var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
        var namesList = IntStream.vary(0, 1000000)
            .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
            .toList();

        // Break up into batches
        int batchSize = 1000;
        CompletableFuture<Map> finalCountsFuture =
            IntStream.iterate(0, batchStart -> batchStart  batchStart + batchSize)
                .mapToObj(batchStart -> prepareBatch(namesList, batchStart, batchSize))
                .cut back(Fundamental::combineFeatures)
                .get();
                
        // Look ahead to the outcome to be computed
        Map finalCounts = finalCountsFuture.be a part of();

        // Discover the max rely
        String mostFrequentName = finalCounts.entrySet()
            .stream()
            .max(Map.Entry.comparingByValue())
            .get()
            .getKey();

        System.out.printf("Essentially the most frequent identify is %spercentn", mostFrequentName);
    }

    non-public static CompletableFuture<Map> combineFeatures(
        CompletableFuture<Map> firstFeature,
        CompletableFuture<Map> secondFeature)
    {
        return firstFeature.thenCombineAsync(secondFeature, Fundamental::mergeCounts);
    }

    non-public static Map mergeCounts(Map stringLongMap, Map stringLongMap2)
    {
        System.out.printf("[%s] Merging counts... n", Thread.currentThread().getName());
        Map accumulator = new HashMap(stringLongMap);
        stringLongMap2.forEach((key, worth) -> accumulator.compute(key, (n, c) -> c == null ? worth : c + worth));
        return accumulator;
    }

    non-public static CompletableFuture<Map> prepareBatch(Record namesList, int batchStart,
                                                                     int batchSize)
    {
        return CompletableFuture.supplyAsync(() -> {
            Map localCounts = new ConcurrentHashMap();
            int batchEnd = Math.min((batchStart + batchSize), namesList.measurement());
            System.out.printf("[%s] Processing batch... n", Thread.currentThread().getName());
            for (String identify: namesList.subList(batchStart, batchEnd))
            {
                localCounts.compute(identify, (n, c) -> c == null ? 1L : c + 1);
            }
            return localCounts;
        });
    }
}

The primary factor to notice on this answer is that finalCounts variable now not represents a shared mutable state. As an alternative, it’s returned from a CompletableFuture#be a part of name. It’s because the CompletbleFeature API permits combining outcomes of a number of options asynchronously through a group of then* strategies (thenCombineAsync on this case). This strategy successfully abstracts away the shared mutable state from a programmer, permitting him to specific what they need to obtain as an alternative of coding how to attain that. Such traits are typical for a useful fashion of programming and monadic API.

Additionally, it must be famous that no specific Executor is outlined. As was stated earlier, CompletableFuture make use of ForkJoinPool by default.

Whereas the offered answer eradicated some drawbacks of the Mesozoic code, it’s nonetheless very fundamental and inconvenient in some points. As an apparent instance, there are not any methods to cut back a group of CompletbleFeature, so you must moreover make use of the Stream API.

6. Current time

Since the usual API didn’t present the required degree of comfort, the group took issues into their very own fingers and developed the Reactive Streams Specification, which is ideologically much like concepts behind the CompletableFuture. Since Java 9 this specification can be represented within the commonplace library by the java.util.concurrent.Move class, which incorporates the java.util.concurrent.Move.Writer and java.util.concurrent.Move.Subscriber interfaces. Nonetheless, it must be clarified that reactive streams are usually not simply improved CompletableFuture. They’re a rethinking of the final strategy to asynchronous computation, which has some similarity to the CompletableFuture API. This might be coated in additional particulars within the subsequent publish.

There are two implementations of the Reactive Streams Specification on the JVM, RxJava and the extra well-liked Venture Reactor. The latter might be utilized in future examples.

Conceptually, Venture Reactor offers two courses with a monadic interface:

• Mono — to course of 0 to 1 components
• Flux — to course of from 0 to infinity components

Mono performance is much like the CompletableFuture however with an incline to the Reactive Streams Specification and far richer API.

Flux could be thought of a java.util.stream.Stream the place the execution of every chain methodology (which is named a reactive operator) could be offloaded to a completely different thread, much like the CompletableFuture.supplyAsync(Provider<U>, Executor) methodology. Flux additionally has wealthy API for asynchronous processing.

With Venture Reactor the answer to the unique drawback will appear to be this:

public class Fundamental
{
    public static void primary(String[] args) throws InterruptedException, ExecutionException
    {
        // Generate names
        Random r = new Random();
        var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
        var namesList = IntStream.vary(0, 1000000)
            .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
            .gather(Collectors.toList());


        int batchSize = 1000;
        var finalCounts = Flux.fromIterable(namesList)
            // Break up to batches
            .buffer(batchSize)
            // Mixture intermediate counts asynchronously
            .flatMap(Fundamental::processBatch)
            .cut back(new HashMap(), Fundamental::mergeIntermediateCount)
            .flatMapIterable(HashMap::entrySet);
        String mostFrequentName = MathFlux.max(finalCounts, Map.Entry.comparingByValue())
            .map(Map.Entry::getKey)
            .block();


        System.out.printf("Essentially the most frequent identify is %spercentn", mostFrequentName);
    }

    non-public static HashMap mergeIntermediateCount(HashMap acc,
                                                                Map intermediateResult)
    {
        intermediateResult.forEach((identify, intermediateCount) -> acc.merge(identify, intermediateCount,
            Lengthy::sum));
        return acc;
    }

    non-public static Mono<Map> processBatch(Record batch)
    {

        return Flux.fromIterable(batch)
            .groupBy(Perform.identification())
            .flatMap(group -> group.rely().map(rely -> Tuples.of(group.key(), rely)))
            .collectMap(Tuple2::getT1, Tuple2::getT2)
            .doOnSubscribe(__ -> System.out.printf("[%s] Processing batch... n", Thread.currentThread().getName()))
            .subscribeOn(Schedulers.boundedElastic());
    }
}

Since Venture Reactor API is designed with frequent operations of asynchronous information processing, it offers all needed performance out of the field, together with batch splitting, splitting the batch into teams, counting, and so forth. It permits code to be uniform and frees a developer from the need to create short-term collections and different crucial constructs. Additionally, this model is multithreaded due to a single line — .subscribeOn(Schedulers.boundedElastic());: it switches the processing of the batch to a devoted thread pool. With out this line, code will nonetheless work, however on a single thread. it is a vital trait of the reactive code: it’s concurrency agnostic.

Reactor, like RxJava, could be thought of to be concurrency-agnostic. That’s, it doesn’t implement a concurrency mannequin. Slightly, it leaves you, the developer, in command. Nonetheless, that doesn’t stop the library from serving to you with concurrency. Reactor docs

All above permits to state that the Reactor API is superior to the CompletableFuture, but it surely comes with critical drawbacks which might be mentioned in later posts.

7. Close to future

Venture Loom is launched in a preview state in JDK 19. Along with the digital threads, it offers an API for the so-called structured concurrency in an incubator state. Structured concurrency introduces the idea of “scope” for asynchronous duties, which lets you congregate asynchronous computation inside this “scope” and construction their execution in a means much like the Java 5 Futures, however with extra capabilities. These scopes, naturally, use digital threads with a purpose to perform submitted duties.

In observe, Loom answer seems to be like this:

public class MainVirtualThreads
{

    public static void primary(String[] args) throws InterruptedException
    {

        Random r = new Random();
        var names = Record.of("Joe", "Monica", "Chandler", "Phoebe", "Rachel", "Ross", "Janice");
        var namesList = IntStream.vary(0, 1000000)
            .mapToObj(__ -> names.get(r.nextInt(names.measurement())))
            .toList();


        strive (var scope = new BatchScope())
        {
            int batchSize = 1000;

            IntStream.iterate(0, batchStart -> batchStart  batchStart + batchSize)
                .mapToObj(batchStart -> prepareBatch(namesList, batchStart, batchSize))
                .forEach(scope::fork);

            scope.be a part of();

            System.out.println("Essentially the most frequent identify is " + scope.mostFrequentName());
        }
    }

    non-public static Callable<Map> prepareBatch(Record namesList, int batchStart, int batchSize)
    {
        return () -> {
            Map localCounts = new ConcurrentHashMap();
            int batchEnd = Math.min((batchStart + batchSize), namesList.measurement());
            System.out.printf("[virtual=%s] Processing batch... n", Thread.currentThread().isVirtual());
            for (String identify: namesList.subList(batchStart, batchEnd))
            {
                localCounts.compute(identify, (n, c) -> c == null ? 1L : c + 1);
            }
            return localCounts;
        };
    }

    non-public static class BatchScope extends StructuredTaskScope<Map>
    {

        non-public last ConcurrentHashMap outcome = new ConcurrentHashMap();

        @Override
        protected void handleComplete(Future<Map> future)
        {
            last var intermediateResult = future.resultNow();
            for (var stringLongEntry: intermediateResult.entrySet())
            {
                outcome.compute(stringLongEntry.getKey(), (n, c) -> updateCount(stringLongEntry.getValue(), c));
            }
        }

        non-public lengthy updateCount(Lengthy newCount, Lengthy existingCount)
        {
            return existingCount == null ? newCount : existingCount + newCount;
        }

        public String mostFrequentName()
        {
            return outcome.entrySet()
                .stream()
                .max(Map.Entry.comparingByValue())
                .get()
                .getKey();
        }
    }
}

As you may see, this strategy is indirectly a return to the Paleozoic strategy, however with modifications and enhancements, which can be according to modern developments and approaches. The brand new API offers a harmonious relationship between asynchronous duties and callbacks to gather the results of their work, in addition to different options, described within the JEP.

This iteration of API evolution clearly demonstrates that the Java architects determined to maneuver in a means that’s extra pure to the JVM platform than Reactive: it depends on a well-known (and verbose) crucial strategy, however with the efficiency of digital threads and the chance to conveniently construction asynchronous duties.

However does this imply the unconditional demise of Reactive? Though some Java architects consider that the reply is sure, additional posts will reveal that there are different prospects.

8. Conclusion

9. Obtain the Supply Code