Spring响应式编程：如何将数据库轮询结果封装为Mono返回

Hey there! I see you're trying to adapt a blocking long-running database polling method into a reactive Mono for your Spring REST endpoint—let's fix this together.

The Problem with Your Current Setup #

Your pollOnDatabase method is blocking and long-running. If you just call it directly inside findFoo and wrap the result in Mono.just(), it will execute immediately on the reactive thread pool (reserved for non-blocking operations). This will block those threads, which kills the scalability of your reactive application.

The Correct Approach #

We need to wrap the blocking operation in an async container and offload it to a dedicated thread pool designed for blocking tasks. Here's how to implement findFoo:

@Service
public class FooService {
    public Mono<Foo> findFoo(String id) {
        // Wrap the blocking poll operation in a Callable to defer execution
        return Mono.fromCallable(() -> pollOnDatabase(id))
                // Offload the blocking task to Reactor's bounded elastic scheduler
                .subscribeOn(Schedulers.boundedElastic());
    }

    private Foo pollOnDatabase(String id) {
        // Your existing long-running polling logic here
    }
}

Key Details Explained #

• Mono.fromCallable(...): This wraps your blocking method so it doesn't execute immediately—instead, it runs only when the Mono is subscribed to (which happens when your REST endpoint receives a request).
• subscribeOn(Schedulers.boundedElastic()): This tells Reactor to run the blocking pollOnDatabase task on a dedicated thread pool for blocking operations. The bounded elastic scheduler dynamically creates threads as needed (with an upper limit) to prevent resource exhaustion.

Additional Tips #

1. Don't use Mono.just(pollOnDatabase(id)): This will run the blocking method synchronously when findFoo is called, which defeats the purpose of reactive programming.
2. Fix your Controller's default case: Uncomment the defaultIfEmpty line to handle cases where pollOnDatabase returns null (or the Mono completes empty):

   @GetMapping("/foo/{id}")
   public Mono<ResponseEntity<Foo>> findById(@PathVariable String id) {
       return fooService.findFoo(id)
               .map(foo -> ResponseEntity.ok(foo))
               .defaultIfEmpty(ResponseEntity.notFound().build());
   }
   

3. Consider reactive database drivers (if possible): If you can replace your blocking JDBC setup with a reactive driver like R2DBC, you can avoid blocking entirely. But if polling is a requirement and you have to use a blocking method, the above approach is the way to go.

内容的提问来源于stack exchange，提问作者user871611