Spring Webflux与Kotlin：在响应式流中正确执行CRUD操作

理解Spring Webflux与响应式编程

spring webflux是spring框架提供的响应式web栈，它基于project reactor库，旨在构建非阻塞、事件驱动的服务。与传统的命令式编程不同，响应式编程的核心在于数据流和变化传播。在webflux应用中，我们通常操作mono（0或1个元素的异步序列）和flux（0到n个元素的异步序列），通过链式操作符来处理数据，而不是立即执行代码。

当处理外部API调用并尝试将结果保存到本地数据库时，一个常见的陷阱是在响应式流的subscribe方法内部执行数据库写入操作。这往往会导致数据未能成功保存，其根本原因在于对响应式流生命周期的误解。

问题分析：为什么在subscribe中执行CRUD会失败

考虑以下场景：一个Spring Webflux服务需要从远程API（如jsonplaceholder）获取数据，然后将这些数据保存到本地PostgreSQL数据库。最初的实现可能如下所示：

@RestController
@RequestMapping("/api")
class AppController(private val appService: AppService) {

    @GetMapping("/jsonplaceholder")
    fun getData(): Mono<ResponseEntity<List<Post>>> {
        val ret =  appService.fetchPosts() // 获取远程数据，返回Flux<Post>
            .take(3) // 取前3条
            .collectList() // 收集为Mono<List<Post>>
            .map { body -> ResponseEntity.ok().body(body) } // 封装为ResponseEntity
            .toMono() // 转换为Mono

        // 问题所在：在subscribe回调中执行数据库写入
        ret.log().subscribe(
            {
                val x:List<Post> = it.body as List<Post>
                for (t in x){
                    print(t)
                    appService.createPost(t) // 调用保存服务
                }
            },null,
            { }
        )
        return ret // 返回响应
    }
}

尽管远程API调用和数据接收看似正常，但数据库中却没有任何数据。这是因为subscribe方法是非阻塞的。当ret.log().subscribe(...)被调用时，它会注册一个回调函数，但并不会等待这个回调函数执行完毕。主线程会立即继续执行并返回ret。

由于数据库保存操作appService.createPost(t)本身也返回一个Mono<Post>，它是一个异步操作。在subscribe回调内部，这些Mono<Post>并没有被“订阅”到，也没有被整合到主响应式流中。这意味着，当HTTP响应已经发送回客户端时，数据库的写入操作可能才刚刚开始，甚至还没有开始。由于Spring Webflux的生命周期管理，一旦主响应式流完成并发出HTTP响应，任何未被正确整合到该流中的异步操作都可能被取消或无法完成。因此，数据库保存操作在大多数情况下会“悄无声息”地失败。

简而言之，subscribe通常用于触发流的执行或处理最终的副作用（如日志记录、更新UI等），而不是在其中执行需要影响主业务流程的异步操作。在响应式编程中，应避免在subscribe内部执行CRUD操作，除非你明确知道这是一个“即发即忘”且不影响HTTP响应的场景。

解决方案：利用flatMap整合异步操作

正确的做法是将数据库保存操作整合到响应式流本身中，而不是将其从流中“剥离”到subscribe回调中。Project Reactor提供了flatMap操作符，它非常适合处理这种场景：当流中的每个元素都需要触发另一个异步操作，并且我们希望将这些异步操作的结果扁平化到主流中时，flatMap是理想选择。

Smart Picture

Smart Picture 智能高效的图片处理工具

77

查看详情

以下是使用flatMap改进后的代码示例：

@RestController
@RequestMapping("/api")
class AppController(private val appService: AppService) {

    @GetMapping("/jsonplaceholder")
    fun getData(): Mono<ResponseEntity<List<Post>>> {
        return appService.fetchPosts() // 获取远程数据，返回Flux<Post>
            .take(3) // 取前3条
            // 核心改变：使用flatMap将每个Post的保存操作整合到流中
            .flatMap { post -> appService.createPost(post) } // 为每个Post调用createPost，返回Mono<Post>
            .collectList() // 收集所有已保存的Post为Mono<List<Post>>
            .map { savedPosts -> ResponseEntity.ok().body(savedPosts) } // 封装为ResponseEntity
            .toMono() // 转换为Mono
    }
}

让我们详细解析这个解决方案：

1. appService.fetchPosts(): 这仍然是获取远程API数据的入口，返回一个Flux<Post>。
2. .take(3): 限制只处理前3个Post对象。
3. .flatMap { post -> appService.createPost(post) }: 这是关键步骤。对于从fetchPosts()流中发出的每个Post对象，flatMap会调用appService.createPost(post)。createPost方法返回一个Mono<Post>，代表一个异步的数据库保存操作。flatMap的作用是将这些独立的Monos“扁平化”回一个单一的Flux<Post>。这意味着，只有当appService.createPost(post)返回的Mono完成（即数据库保存成功）后，下一个元素才会继续处理，并且这个保存操作的结果会被传递到下游。
4. .collectList(): 在所有Post都经过flatMap处理（即保存到数据库）之后，将它们收集到一个List<Post>中，并封装在一个Mono<List<Post>>中。
5. .map { savedPosts -> ResponseEntity.ok().body(savedPosts) }: 将最终保存的Post列表封装成一个ResponseEntity。
6. .toMono(): 确保最终返回类型符合控制器方法签名。

通过这种方式，数据库保存操作被完全集成到响应式流中。整个链条是原子性的，只有当所有数据库操作都完成后，collectList才会发出结果，进而触发map操作，最终HTTP响应才会被发送。这保证了数据持久化的正确执行。

最佳实践与注意事项

• 避免在subscribe中执行核心业务逻辑：subscribe是流的终结操作，通常用于触发流、日志记录或在流完成时执行一些最终清理。核心的业务逻辑（如数据转换、验证、数据库操作、外部服务调用）应该使用操作符（如map, flatMap, filter, zip等）来构建。
• 理解map与flatMap的区别：
  • map用于同步地转换流中的元素，它接收一个返回非Publisher类型（如Post）的函数。
  • flatMap用于异步地转换流中的元素，它接收一个返回Publisher类型（如Mono<Post>或Flux<Post>）的函数，并将这些内部Publisher的结果扁平化到主Publisher中。当操作涉及I/O（如数据库访问、网络请求）时，通常需要使用flatMap。
• 错误处理：在响应式流中，错误会沿流传播。可以使用onErrorResume, retry, doOnError等操作符来处理错误，确保应用的健壮性。
• 事务管理：对于R2DBC，事务管理通常通过TransactionalOperator或Spring的 @Transactional注解（配合ReactiveTransactionManager）来实现。确保跨多个数据库操作的原子性。
• 日志记录：log()操作符在开发和调试阶段非常有用，可以清晰地看到流中事件的传播。但在生产环境中，应谨慎使用，或配置更精细的日志级别。

总结

在使用Spring Webflux和Kotlin构建响应式应用时，正确处理异步操作（尤其是涉及数据库I/O的CRUD操作）至关重要。将数据库写入等副作用操作集成到响应式流中，利用flatMap等操作符进行链式调用，是确保数据持久化和维护非阻塞特性的关键。避免在subscribe回调中执行核心业务逻辑，有助于构建更健壮、更符合响应式编程范式的应用程序。

以上就是Spring Webflux与Kotlin：在响应式流中正确执行CRUD操作的详细内容，更多请关注php中文网其它相关文章！