Understanding the Reactive Thread Model: Part 1

Reactive or non-blocking processing is in high demand, but before adopting it, one should deeply understand its thread model. For thread model two things are very important to know: thread communication and execution flow. In this blog, I will try to explain both of these topics in-depth.

What Is Reactive Programming?

There are lots of definitions on the web; the Wiki definition is a bit theoretical and generic. From a threading perspective, my version is "Reactive Programming is the processing of the asynchronous event stream, on which you can observe.”                                                                    

You can find much more discussion about Reactive Programming on the web but for now, let’s stick to our topic of the Reactive Thread Model. Let’s start with a very simple reactive use case, where we want to return the sum of an integer array. 

Our main request thread should not get blocked while processing the sum of an integer array. Let’s start by creating a simple WebServer.

xxxxxxxxxx

        ServerSocket server = new ServerSocket(9090);

        while (true) { 

            try (Socket socket = server.accept()) {

                Consumer<Integer> response = a -> {

                    String responseStr  = "HTTP/1.1 200 OK\r\n\r\n"+"Result= "+ a + "                                           and Thread: "+Thread.currentThread();

                    try {

                        socket.getOutputStream().write(responseStr.getBytes("UTF-8"));

                    } catch (IOException e) {

                        e.printStackTrace();

                    }

                };

                Random random = new Random();

                ctx.getBean(ReactorComponent.class).nonBlockingSum(new Integer[] {

                        random.nextInt(), random.nextInt(), random.nextInt()

                }).subscribe(response);

                TimeUnit.MILLISECONDS.sleep(200);

        } 

Here, we are creating a socket server, opening a socket, and keeping the socket alive until the asynchronous processing is completed. Asynchronous processing is happening by calling the nonBlockingSum method and passing the consumer function or lambda as an observable. Once the sum is ready, our function/lambda will get a callback. From the callback, we return the sum value to the client via a socket.

So, if you call the URL, http://localhost:9090 in parallel/sequence, you will get the following response:

xxxxxxxxxx

Result= 83903382 and Thread: Thread[ReactiveScheduler-2,5,main]

Result= -1908131554 and Thread: Thread[ReactiveScheduler-3,5,main]

The above is just used as an example. In the real world, you should use netty/undertow/servlet 3.1 as the reactive webserver. Now let’s get somewhat deep and try to understand the following flows:

1. Blocking Call.
2. Non-blocking call.
3. Non-blocking call with thread execution.
4. Serial business flow processing.
5. Parallel business flow processing.

We are going to use Spring WebFlux, which is built on top of the Reactor framework for Reactive programming. Let’s cover sections 1 and 2 in this blog and other sections in part-2 so that it will be very easy to understand.

We are going to write a simple sum method and make it Reactive using the supplier function.

xxxxxxxxxx

public Integer getSum(final Integer arr[]) {

        Integer count = 0;

        for (int i = 0; i < arr.length; i++) {

            count += arr[i];

        }

        return count;

}

xxxxxxxxxx

public Mono<Integer> nonBlockingSum(final Integer arr[]) throws InterruptedException {

    Mono<Integer> m = Mono.fromSupplier(() ->             

                    this.computationService.getSum(arr)).subscribeOn(this.scheduler);

        return m;

}

Blocking Call

xxxxxxxxxx

Integer t = ctx.getBean(ReactorComponent.class).nonBlockingSum(new Integer[] {4,78,4,676,3,45}).block();

As shown in the diagram, the request thread is getting blocked until the computation of the sum is completed. If we execute the code, we will get the following response:

xxxxxxxxxx

In ReactiveApplication.blockingCall: Thread[main,5,main]

In ReactorComponent.nonBlockingSum: Thread[main,5,main]

Returning form ReactorComponent.nonBlockingSum: Thread[main,5,main]

In ComputationService.getSum: Thread[ReactiveScheduler-2,5,main]

Returning from ComputationService.getSum: Thread[ReactiveScheduler-2,5,main]

Returning from ReactiveApplication.blockingCall result= 810

This clearly shows that the blocking call waited until the sum execution was completed.

Non-Blocking Call

xxxxxxxxxx

public static void nonBlockingCall(ApplicationContext ctx) throws InterruptedException{

        Consumer<Integer> display = a -> {

            System.out.println("In Consumer/Lambada result= "+ a + " and Thread: 

                                "+Thread.currentThread());

        };

        ctx.getBean(ReactorComponent.class).nonBlockingSum(new Integer[] 

                                    {4,78,4,676,3,45}).subscribe(display);

}

Here, the request thread is not blocked, and the execution of the sum is shifted to a thread allocated from the thread pool. The callback and function/lambda are also executed on the same thread. If we execute, the code, we will get the following response:

xxxxxxxxxx

In ReactiveApplication.nonBlockingCall: Thread[main,5,main]

In ReactorComponent.nonBlockingSum: Thread[main,5,main]

Returning form ReactorComponent.nonBlockingSum: Thread[main,5,main]

Returning from ReactiveApplication.nonBlockingCall: Thread[main,5,main]

In ComputationService.getSum: Thread[ReactiveScheduler-2,5,main]

Returning from ComputationService.getSum: Thread[ReactiveScheduler-2,5,main]

In Consumer/Lambada result= 810 and Thread: Thread[ReactiveScheduler-2,5,main]

This clearly shows that the request thread didn’t wait until the sum is computed. Also, the consumer and sum were processed in the same thread.

Sections 3, 4, and 5 will be covered in part 2, and you can get the code on GitHub.