Reactor WebFlux vs Virtual Threads

1. Overview

In this tutorial, we’ll compare Java 19’s virtual threads with Project Reactor’s Webflux. We’ll begin by revisiting the fundamental workings of each approach, and subsequently, we’ll analyze their strengths and weaknesses.

We’ll start by exploring the strengths of reactive frameworks and we’ll see why WebFlux remains valuable. After that, we’ll discuss the thread-per-request approach and highlight scenarios where virtual threads can be a better option.

2. Code Examples

For the code examples in this article, we’ll assume we’re developing the backend of an e-commerce application. We’ll focus on the function responsible for computing and publishing the price of an item added to a shopping cart:

class ProductService {
    private final String PRODUCT_ADDED_TO_CART_TOPIC = "product-added-to-cart";

    private final ProductRepository repository;
    private final DiscountService discountService;
    private final KafkaTemplate<String, ProductAddedToCartEvent> kafkaTemplate;

    // constructor

    public void addProductToCart(String productId, String cartId) {
        Product product = repository.findById(productId)
          .orElseThrow(() -> new IllegalArgumentException("not found!"));

        Price price = product.basePrice();
        if (product.category().isEligibleForDiscount()) {
            BigDecimal discount = discountService.discountForProduct(productId);
            price.setValue(price.getValue().subtract(discount));
        }

        var event = new ProductAddedToCartEvent(productId, price.getValue(), price.getCurrency(), cartId);
        kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event);
    }

}

As we can see, we begin by retrieving the Product from the MongoDB database using a MongoRepository. Once retrieved, we determine if the Product qualifies for discounts. If this is the case, we use DiscountService to perform an HTTP request to ascertain any available discounts for the product.

Finally, we calculate the final price for the product. Upon completion, we dispatch a Kafka message containing the productId, cartId, and the computed price.

3. WebFlux

WebFlux is a framework for building asynchronous, non-blocking, and event-driven applications. It operates on reactive programming principles, leveraging the Flux and Mono types to handle the intricacies of asynchronous communication. These types implement the publisher-subscriber design pattern, decoupling the consumer and the producer of the data.

3.1. Reactive Libraries

Numerous modules from the Spring ecosystem integrate with WebFlux for reactive programming. Let’s use some of these modules while refactoring our code toward a reactive paradigm.

For instance, we can switch the MongoRepository to a ReactiveMongoRepository. This change means we’ll have to work with a Mono<Product> instead of an Optional<Product>:

Mono<Product> product = repository.findById(productId)
  .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("not found!")));

Similarly, we can change the ProductService to be asynchronous and non-blocking. For example, we can make it use WebClient for performing the HTTP requests, and, consequently, return the discount as a Mono<BigDecimal>:

Mono<BigDecimal> discount = discountService.discountForProduct(productId);

3.2. Immutability

In functional and reactive programming paradigms, immutability is always preferred over mutable data. Our initial method involves altering the Price‘s value using a setter. However, as we move towards a reactive approach, let’s refactor the Price object and make it immutable.

For example, we can introduce a dedicated method that applies the discount and generates a new Price instance rather than modifying the existing one:

record Price(BigDecimal value, String currency) {  
    public Price applyDiscount(BigDecimal discount) {
        return new Price(value.subtract(discount), currency);
    }
}

Now, we can compute the new price based on the discount, using WebFlux’s map() method:

Mono<Price> price = discountService.discountForProduct(productId)
  .map(discount -> price.applyDiscount(discount));

Additionally, we can even use a method reference here, to keep the code compact:

Mono<Price> price = discountService.discountForProduct(productId).map(price::applyDiscount);

3.3. Functional Pipelines

Mono and Flux adhere to the functor and monad patterns, through methods such as map() and flatMap(). This allows us to describe our use case as a pipeline of transformations on immutable data.

Let’s try to identify the transformations needed for our use case:

• we start with a raw productId
• we transform it into a Product
• we use the Product to compute a Price
• we use the Price to create an event
• finally, we publish the event on a message queue

Now, let’s refactor the code to reflect this chain of functions:

void addProductToCart(String productId, String cartId) {
    Mono<Product> productMono = repository.findById(productId)
      .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("not found!")));

    Mono<Price> priceMono = productMono.flatMap(product -> {
        if (product.category().isEligibleForDiscount()) {
            return discountService.discountForProduct(productId)
              .map(product.basePrice()::applyDiscount);
        }
        return Mono.just(product.basePrice());
    });

    Mono<ProductAddedToCartEvent> eventMono = priceMono.map(
      price -> new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId));

    eventMono.subscribe(event -> kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event));
}

Now, let’s inline the local variables to keep the code compact. Additionally, let’s extract a function for computing the price, and use it inside of the flatMap():

void addProductToCart(String productId, String cartId) {
    repository.findById(productId)
      .switchIfEmpty(Mono.error(() -> new IllegalArgumentException("not found!")))
      .flatMap(this::computePrice)
      .map(price -> new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId))
      .subscribe(event -> kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event));
}

Mono<Price> computePrice(Product product) {
    if (product.category().isEligibleForDiscount()) {
        return discountService.discountForProduct(product.id())
          .map(product.basePrice()::applyDiscount);
    }
    return Mono.just(product.basePrice());
}

4. Virtual Threads

Virtual Threads were introduced in Java via Project Loom as an alternative solution for parallel processing. They are lightweight, user-mode threads managed by the Java Virtual Machine (JVM). As a result, they are particularly well suited for I/O operations, where traditional threads may spend significant time waiting for external resources.

In contrast to asynchronous or reactive solutions, virtual threads enable us to keep using the thread-per-request processing model. In other words, we can keep writing code sequentially, without mixing the business logic and the reactive API.

4.1. Virtual Threads

There are several approaches available to utilize virtual threads for executing our code. For a single method, such as the one demonstrated in the previous example, we can employ startVirtualThread(). This static method was recently added to the Thread API and executes a Runnable on a new virtual thread:

public void addProductToCart(String productId, String cartId) {
    Thread.startVirtualThread(() -> computePriceAndPublishMessage(productId, cartId));
}

private void computePriceAndPublishMessage(String productId, String cartId) {
    // ...
}

Alternatively, we can create an ExecutorService that relies on virtual threads with the new static factory method Executors.newVirtualThreadPerTaskExecutor(). Furthermore, for applications using Spring Framework 6 and Spring Boot 3, we can leverage the new Executor and configure Spring to favor virtual threads over platform threads.

4.2. Compatibility

Virtual threads simplify code by using a more traditional synchronous programming model. As a result, we can write code in a sequential manner, akin to blocking I/O operations, without worrying about explicit reactive constructs.

Moreover, we can seamlessly switch from regular single-threaded code to virtual threads with minimal to no alterations. For instance, in our previous example, we simply need to create a virtual thread using the static factory method startVirtualThread() and execute logic inside of it:

void addProductToCart(String productId, String cartId) {
    Thread.startVirtualThread(() -> computePriceAndPublishMessage(productId, cartId));
}

void computePriceAndPublishMessage(String productId, String cartId) {
    Product product = repository.findById(productId)
      .orElseThrow(() -> new IllegalArgumentException("not found!"));

    Price price = computePrice(productId, product);

    var event = new ProductAddedToCartEvent(productId, price.value(), price.currency(), cartId);
    kafkaTemplate.send(PRODUCT_ADDED_TO_CART_TOPIC, cartId, event);
}

Price computePrice(String productId, Product product) {
    if (product.category().isEligibleForDiscount()) {
        BigDecimal discount = discountService.discountForProduct(productId);
        return product.basePrice().applyDiscount(discount);
    }
    return product.basePrice();
}

4.3. Readability

With the thread-per-request processing model, it can be easier to understand and reason about the business logic. This can reduce the cognitive load associated with reactive programming paradigms.

In other words, virtual threads allow us to cleanly separate the technical concerns from our business logic. As a result, it eliminates the need for external APIs in implementing our business use cases.

5. Conclusion

In this article, we compared two different approaches to concurrency and asynchronous processing. We started by analyzing the project Reactor’s WebFlux and the reactive programming paradigm. We discovered that this approach favors immutable objects and functional pipelines.

After that, we discussed virtual threads and their exceptional compatibility with legacy codebases that allow for a smooth transition to non-blocking code. Additionally, they have the added benefit of cleanly separating the business logic from the infrastructure code and other technical concerns.