Multithread execution on Spring 4 using Reactor 2.0

Question

I'm trying to integrate Reactor 2x into an existing Spring 4 application to boost performance during the execution of a REST request where resources can be fetched independent from each other, sort of map-reduce where we parallelize a job into multiple threads and then join them into a buffer.

So far we have this sample working in a non-spring environment:

    //Ordered resources to apply transformations.
    List<Map<String, Object>> result;

    result = Streams.from(resources)
                    .flatMap(m -> Streams.just(m) .dispatchOn(Environment.cachedDispatcher())
                    .map(resourceToMapFunction::apply))
                    .buffer().next().await(5, TimeUnit.SECONDS);

In the sample above, we apply the transformation using the resourceToMapFunction and then the join the with the buffer() method, create the Promise to wait for the result and return the result.

My first question, is this the way Reactor is suppose to be used? I know that the transformations are applied correctly but maybe, me being a newbie in Reactor not using something the right way.

My second question, is not a big deal but is there anything in the Reactor project to return in the same order provided in the resources input? Since this is executing in multiple threads I'm pretty sure the answer is not, and like I said is the less of my worries, but wanted to ask anyways.

Last question, when I introduce this code to my Spring project the transformations failed because the Bean dependencies to apply the underlaying transformations are not in the execution threads, is this something I can do easily with the Spring Reactor version? If so, is there any link or document that shows how to do it?

Thanks a lot!

José Luis

Answer 1

You should use Reactor's API. So instead of creating a Java Stream you should create Reactor Flux. Then you should use the flatMap function of a Flux in order to fetch some resources. If you want the resources to be in the same order as requested you can use flatMapSequential.

So the code would look like:

        Flux<Map<String,Object>> result = Flux.fromIterable(resources)
                .flatMapSequential(resourceToMapFunction::apply)
                .take(Duration.ofSeconds(5));

Then you have a Reactor stream and you can apply more operations to that stream. If you'd like to have simple list then you would have to block by using operator collectList. If you would like to have a control over a Thread on which given function will be executed then you'd have to get familiar with Scheduler.