Spring Boot Asynchronous Processing and Reactive Programming

1. Asynchronous Processing in Spring Boot

What is Asynchronous Processing?

Asynchronous processing allows a program to handle multiple tasks concurrently without blocking the main thread. Instead of waiting for a task to complete before moving to the next one, the program can initiate a task and continue executing other operations, then handle the result when it becomes available.

Key Benefits:

• Improved performance: Better utilization of system resources

• Better scalability: Handle more concurrent requests

• Responsive applications: Main thread remains responsive

Spring Boot Implementation:

Spring provides @Async annotation for simple asynchronous method execution.

java

@Service
public class NotificationService {
    
    @Async  // This method will be executed asynchronously
    public CompletableFuture<String> sendNotification(String message) {
        // Simulate time-consuming task
        Thread.sleep(2000);
        return CompletableFuture.completedFuture("Notification sent: " + message);
    }
}

2. Reactive Programming with Project Reactor (Mono & Flux)

What is Reactive Programming?

Reactive programming is a programming paradigm focused on asynchronous data streams and the propagation of change. It's particularly useful for:

• High-concurrency applications

• Real-time systems

• Applications with streaming data

Core Principles (Reactive Manifesto):

• Responsive: Systems respond in a timely manner

• Resilient: Systems stay responsive in the face of failure

• Elastic: Systems stay responsive under varying workload

• Message-driven: Rely on asynchronous message-passing

3. Mono and Flux Explained

Mono: Zero or One Element

Definition: A Mono is a reactive type that represents a stream which can emit 0 or 1 element, then complete. It's similar to a Future or Optional but with reactive capabilities.

Real-world Example: Fetching a single user profile from a database

• Success case: Returns one user object

• Error case: Returns error (no user)

• Empty case: Returns empty (user not found)

Flux: Zero to N Elements

Definition: A Flux is a reactive type that represents a stream which can emit 0 to N elements, then complete. It's similar to a List or Collection but with reactive streaming capabilities.

Real-world Example: A live sports score feed

• Multiple score updates over time

• Continuous stream of data

• Can complete when the game ends

4. Code Examples with Detailed Explanations

Example 1: Mono - Single User Retrieval

Scenario: Fetching a user from a database. We want to handle this reactively to avoid blocking threads.

java

import reactor.core.publisher.Mono;

@Service
public class UserService {
    
    // Simulating a repository call
    public Mono<User> findUserById(String userId) {
        return Mono.fromCallable(() -> {
            // Simulate database call
            Thread.sleep(100); // Simulate latency
            
            if ("valid-id".equals(userId)) {
                return new User(userId, "John Doe", "john@example.com");
            } else if ("empty-id".equals(userId)) {
                return null; // Will be converted to Mono.empty()
            } else {
                throw new RuntimeException("Database error");
            }
        });
    }
}

// Controller
@RestController
@RequestMapping("/users")
public class UserController {
    
    @Autowired
    private UserService userService;
    
    @GetMapping("/{id}")
    public Mono<ResponseEntity<User>> getUser(@PathVariable String id) {
        return userService.findUserById(id)
            .map(user -> ResponseEntity.ok(user))
            .defaultIfEmpty(ResponseEntity.notFound().build())
            .onErrorResume(error -> 
                Mono.just(ResponseEntity.status(500).build())
            );
    }
}

Example 2: Flux - Real-time Stock Ticker

Scenario: Streaming real-time stock prices to clients. Each price update is an event in the stream.

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import java.time.Duration;

@Service
public class StockTickerService {
    
    // Create a multicast sink for real-time updates
    private final Sinks.Many<StockPrice> stockPriceSink = 
        Sinks.many().multicast().onBackpressureBuffer();
    
    // Simulate stock price updates
    public void startStockUpdates() {
        Flux.interval(Duration.ofSeconds(1))
            .map(tick -> generateRandomStockPrice())
            .subscribe(stockPriceSink::tryEmitNext);
    }
    
    private StockPrice generateRandomStockPrice() {
        String[] symbols = {"AAPL", "GOOGL", "MSFT", "AMZN"};
        String symbol = symbols[(int)(Math.random() * symbols.length)];
        double price = 100 + (Math.random() * 1000);
        return new StockPrice(symbol, price);
    }
    
    // Stream for a specific stock symbol
    public Flux<StockPrice> getStockStream(String symbol) {
        return stockPriceSink.asFlux()
            .filter(stock -> stock.getSymbol().equals(symbol))
            .doOnSubscribe(sub -> 
                System.out.println("New subscriber for " + symbol)
            );
    }
}

// WebSocket Controller for real-time updates
@RestController
@RequestMapping("/stocks")
public class StockController {
    
    @Autowired
    private StockTickerService stockService;
    
    @GetMapping(value = "/stream/{symbol}", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<StockPrice> streamStockPrices(@PathVariable String symbol) {
        return stockService.getStockStream(symbol)
            .delayElements(Duration.ofSeconds(1)) // Throttle updates
            .doOnNext(price -> 
                System.out.println("Sending: " + price)
            );
    }
}

Example 3: Combining Mono and Flux - E-commerce Order Processing

Scenario: Processing an order and its items reactively.

java

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Service
public class OrderService {
    
    // Mono: Creating a single order
    public Mono<Order> createOrder(OrderRequest request) {
        return Mono.fromCallable(() -> {
            // Validate and create order
            Order order = new Order();
            order.setOrderId(UUID.randomUUID().toString());
            order.setCustomerId(request.getCustomerId());
            return order;
        });
    }
    
    // Flux: Processing multiple order items
    public Flux<OrderItem> processOrderItems(String orderId, List<ItemRequest> items) {
        return Flux.fromIterable(items)
            .flatMap(item -> processSingleItem(orderId, item))
            .doOnNext(processedItem -> 
                System.out.println("Processed: " + processedItem)
            )
            .doOnComplete(() -> 
                System.out.println("All items processed for order: " + orderId)
            );
    }
    
    // Mono: Processing single item
    private Mono<OrderItem> processSingleItem(String orderId, ItemRequest item) {
        return Mono.fromCallable(() -> {
            // Simulate item processing
            Thread.sleep(100); // Simulate processing time
            
            OrderItem orderItem = new OrderItem();
            orderItem.setItemId(UUID.randomUUID().toString());
            orderItem.setProductId(item.getProductId());
            orderItem.setQuantity(item.getQuantity());
            orderItem.setOrderId(orderId);
            
            return orderItem;
        });
    }
    
    // Combining Mono and Flux
    public Mono<OrderSummary> processCompleteOrder(OrderRequest request) {
        return createOrder(request)
            .flatMap(order -> {
                // Process items after order is created
                return processOrderItems(order.getOrderId(), request.getItems())
                    .collectList()  // Convert Flux to Mono<List>
                    .map(items -> {
                        OrderSummary summary = new OrderSummary();
                        summary.setOrder(order);
                        summary.setItems(items);
                        summary.setTotalItems(items.size());
                        return summary;
                    });
            });
    }
}

5. Key Operators and Patterns

Common Mono Operators:

java

// Transforming
Mono<String> mono = Mono.just("hello");
Mono<Integer> lengthMono = mono.map(String::length);

// Handling errors
Mono<User> safeMono = getUser()
    .onErrorReturn(fallbackUser)
    .onErrorResume(error -> getBackupUser());

// Combining Monos
Mono<User> userMono = getUser();
Mono<Profile> profileMono = getProfile();
Mono<UserProfile> combined = userMono.zipWith(profileMono, UserProfile::new);

Common Flux Operators:

java

// Filtering and transforming
Flux<Integer> numbers = Flux.range(1, 10);
Flux<Integer> evenNumbers = numbers.filter(n -> n % 2 == 0);

// Buffering
Flux<List<Integer>> buffered = numbers.buffer(3);

// Windowing (similar to buffering but returns Flux of Flux)
Flux<Flux<Integer>> windowed = numbers.window(3);

// Backpressure handling
Flux<Integer> controlled = numbers.onBackpressureBuffer(100);

6. Best Practices

When to Use Asynchronous (@Async):

• Simple background tasks

• Email sending

• File processing

• When you don't need complex stream processing

When to Use Reactive (Mono/Flux):

• High-concurrency APIs

• Real-time streaming

• Microservices communication

• When you need backpressure management

• Chat applications, live feeds, IoT data streams

Performance Considerations:

1. Cold vs Hot Publishers:

   • Cold: Starts emitting on subscription (database queries)

   • Hot: Emits regardless of subscribers (stock tickers)

2. Backpressure: Always consider how to handle when producer is faster than consumer

3. Resource Management: Use doOnSubscribe, doOnCancel, and doFinally for cleanup

7. Testing Reactive Code

java

@Test
public void testMonoUser() {
    StepVerifier.create(userService.findUserById("valid-id"))
        .expectNextMatches(user -> 
            user.getName().equals("John Doe")
        )
        .verifyComplete();
}

@Test
public void testFluxStockStream() {
    StepVerifier.withVirtualTime(() -> 
            stockService.getStockStream("AAPL").take(3)
        )
        .thenAwait(Duration.ofSeconds(3))
        .expectNextCount(3)
        .verifyComplete();
}

Summary

Asynchronous Processing (@Async) is simpler and suitable for basic background tasks, while Reactive Programming (Mono/Flux) provides more powerful abstractions for handling streams of data with built-in backpressure and error handling.

Choose Mono when dealing with single or optional results (like API responses, single database records). Choose Flux when working with collections, streams, or multiple events over time (like WebSocket messages, file streams, database result sets).

The reactive approach shines in microservices architectures and real-time applications where non-blocking I/O and efficient resource utilization are critical.