Avoiding Anti-Patterns in Reactive Programming with Java

Reactive programming is a paradigm that deals with asynchronous data streams and the propagation of change. It is a powerful tool that can simplify complex code, making it easier to read, maintain, and debug. However, like any other tool, it can be misused, leading to ‘anti-patterns’. In this post, we will explore some common anti-patterns in reactive programming and how to avoid them.

Photo by Peter Livesey on Unsplash

1. Blocking Calls

One of the most common anti-patterns in reactive programming is the use of blocking calls. Reactive programming is designed to be non-blocking to allow for more efficient use of resources. However, when a blocking call is used, it defeats the purpose of using reactive programming.

/** Anti-Pattern **/
private Flux<Orders> getUserOrders(String userId) {
  User user = externalApi.getUser(userId); //blocking call
  List<Order> orders = externalApi.getOrders(user.getId()); //blocking call
  return Flux.fromIterable(orders);
}

In this example, getUser() and getOrders() are blocking calls. Each will block the execution until they complete. Instead of making blocking calls, you can use Project Reactor’s Mono.fromCallable() or Mono.fromSupplier() to wrap these calls and make them non-blocking.

/** Pattern **/
private Flux<Order> getUserOrders(String userId) {
    return Mono.fromCallable(() -> externalApi.getUser(userId))
        .flatMapMany(user -> Mono.fromCallable(() -> externalApi.getOrders(user.getId()))
        .flatMapIterable(orders -> orders));
}

2. Using Incorrect Scheduler

Schedulers control the threading model in reactive programming. Using the incorrect scheduler can lead to performance issues and unexpected behavior.

/** Anti-Pattern **/
Mono.fromCallable(this::someBlockingMethod)
    .subscribeOn(Schedulers.parallel())
    .subscribe();

In this example, someBlockingMethod() is a blocking call. By using Schedulers.parallel(), we’re potentially blocking one of the few threads available in the parallel scheduler, which can lead to performance issues. For blocking calls, use Schedulers.boundedElastic() .

/** Pattern **/
Mono.fromCallable(this::someBlockingMethod)
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe();

Schedulers determine the thread on which the execution of a certain task will happen. There are several types of schedulers available in Project Reactor:

• Schedulers.immediate(): Runs tasks on the current thread. If a task is scheduled from a different thread, it will block that thread.
• Schedulers.single(): Reuses a single thread for all scheduled tasks. It’s useful for tasks that need to be executed in a sequential manner.
• Schedulers.elastic(): Creates a new thread for each task, reusing idle threads when possible. It’s useful for blocking tasks.
• Schedulers.parallel(): Uses a fixed pool of workers, making it suitable for parallel work. It’s not suitable for tasks that block.
• Schedulers.boundedElastic(): Creates new worker threads as needed, but reuses idle ones. It’s useful for both non-blocking and blocking tasks.

3. Overuse of Subscribe()

The subscribe() method triggers the execution of reactive streams. However, overusing this method can lead to uncontrolled execution and resource management issues.

/** Anti-Pattern **/
someFlux.map(item -> {
    anotherFlux.subscribe();
    return item;
}).subscribe();

In this example, anotherFlux.subscribe() is called for each item in someFlux. This can lead to a large number of active subscriptions and potential memory leaks. Instead of subscribing inside a map operator, you can use flatMap to handle inner publishers.

/** Pattern **/
someFlux.flatMap(item -> anotherFlux)
    .subscribe();

4. Ignoring Backpressure

Backpressure is a mechanism that allows consumers to signal to producers how much data they can handle at a time. Ignoring backpressure can lead to overwhelming the consumer and potentially crashing the application.

/** Anti-Pattern **/
Flux.range(1, 1_000_000)
    .subscribe(System.out::println);

In this example, a large range of numbers is created and immediately printed out. There’s no backpressure handling in place, so if the consumer can’t keep up with the producer, it could lead to problems. You can use the onBackpressure* operators to handle backpressure.

/** Pattern **/
Flux.range(1, 1_000_000)
    .onBackpressureBuffer(1024)
    .subscribe(System.out::println);

5. Mixing Reactive and Non-Reactive Code

Mixing reactive and non-reactive code can cause unexpected behavior and thread blocking.

/** Anti-Pattern **/
Mono.just(someMethod())
    .subscribeOn(Schedulers.elastic())
    .block();

In this example, someMethod() is likely a non-reactive method. By using block(), you’re forcing reactive code to behave in a non-reactive way. Instead, you can transform your non-reactive code to reactive using Mono.fromCallable() or Mono.fromSupplier().

/** Pattern **/
Mono.fromCallable(() -> someMethod())
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe();

6. Improper Error Handling

Reactive programming provides a robust error handling mechanism, but it can be misused or ignored.

/** Anti-Pattern **/
Flux.just("1", "2", "three", "4", "5")
    .map(Integer::parseInt)
    .subscribe(System.out::println);

In this example, an error will occur when trying to parse “three” as an integer. However, there’s no error handling logic in place, so the application will crash. You can use the onError* operators to handle errors gracefully.

/** Pattern **/
Flux.just("1", "2", "three", "4", "5")
    .map(Integer::parseInt)
    .onErrorContinue((throwable, o) -> System.out.println("Error parsing: " + o))
    .subscribe(System.out::println);

Reactive programming in Java provides a powerful way to handle asynchronous data streams. However, it’s crucial to understand and avoid these common pitfalls to ensure your code remains efficient and maintainable. Happy Coding!