Spring Boot WebFlux下基于Jackson的响应式YAML文件读取实现方案

原实现中使用ResourceUtils.openStream()读取文件、ObjectMapper.readValue()基于阻塞流解析的操作均属于阻塞IO，会占用线程等待资源返回，不符合响应式非阻塞的规范，可按以下步骤改造：

1. 原有YAML序列化Bean可直接复用 #

原有yamlObjectMapper的Bean逻辑不需要调整，直接保留即可：

@Bean
public ObjectMapper yamlObjectMapper() {
    ObjectMapper yamlObjectMapper = new ObjectMapper(new YAMLFactory().disable(YAMLGenerator.Feature
        .WRITE_DOC_START_MARKER));
    yamlObjectMapper.findAndRegisterModules();
    return yamlObjectMapper;
}

2. 改造CustomerService为响应式实现 #

核心调整点为：返回值改为响应式发布者Mono<Customer>，用异步非阻塞的方式读取文件，将反序列化操作调度到弹性线程池避免阻塞事件循环：

import lombok.Data;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import org.springframework.core.io.ResourceLoader;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.core.io.buffer.DefaultDataBufferFactory;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;
import java.io.IOException;
import java.nio.channels.AsynchronousFileChannel;
import java.nio.file.StandardOpenOption;

@Service
public class CustomerService {      

    @Autowired
    @Qualifier("yamlObjectMapper")
    private ObjectMapper yamlObjectMapper;

    @Autowired
    private ResourceLoader resourceLoader;

    public Mono<Customer> get() {
        // 加载资源后异步非阻塞读取文件内容
        return Mono.fromCallable(() -> resourceLoader.getResource("classpath:/files/test.yaml"))
                .flatMap(resource -> Mono.fromCallable(() -> 
                        AsynchronousFileChannel.open(resource.getFile().toPath(), StandardOpenOption.READ))
                        .flatMap(channel -> DataBufferUtils.readAsynchronousFileChannel(
                                () -> channel, 
                                DefaultDataBufferFactory.sharedInstance, 
                                4096
                        )
                        // 合并所有读取到的缓冲区
                        .reduce(DataBuffer::write)
                        .map(dataBuffer -> {
                            byte[] content = new byte[dataBuffer.readableByteCount()];
                            dataBuffer.read(content);
                            // 释放缓冲区避免内存泄漏
                            DataBufferUtils.release(dataBuffer);
                            return content;
                        }))
                )
                // 将反序列化操作调度到弹性线程池执行，避免阻塞事件循环线程
                .publishOn(Schedulers.boundedElastic())
                .map(content -> yamlObjectMapper.readValue(content, Customer.class))
                // 统一异常转换
                .onErrorMap(IOException.class, ex -> new IllegalStateException("读取或解析YAML文件失败", ex));
    }

    @Data
    public static class Customer {
        private String name;
        private String surname;
        private String email;
    }
}

核心改造说明 #

• 所有IO操作都采用异步非阻塞实现，不会让线程陷入空等状态，符合响应式编程的非阻塞要求
• 耗时的反序列化操作单独调度到Schedulers.boundedElastic()弹性线程池执行，不会阻塞响应式核心的事件循环线程
• 异常处理完全融入响应式流，不需要手动捕获抛出异常，符合Reactor的API规范
• 返回值为Mono<Customer>，可直接接入Spring WebFlux等响应式技术栈使用

注意：如果项目还未引入响应式相关依赖，Spring Boot项目可直接引入spring-boot-starter-webflux依赖，非Spring Boot项目需引入对应版本的reactor-core和spring-core依赖。

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