Spring Boot Webflux中S3异步下载致下游服务调用延迟问题

Spring Boot Webflux应用原本通过WebClient并行调用5个下游服务正常，但使用Spring调度器结合aws-s3-async SDK定期从S3读取文件时，会导致下游服务调用延迟上升，进而引发整体服务延迟连锁反应。

WebClient配置 #

ConnectionProvider connectionProvider = ConnectionProvider
                .builder("webclient-conn-pool")
                .maxConnections(1000)
                .maxIdleTime(Duration.ofMillis(75000))
                .maxLifeTime(Duration.ofMillis(150000))
                .metrics(true)
                .pendingAcquireMaxCount(httpConnPoolProperties.getPendingAcquireMaxCount())
                //.evictInBackground(Duration.ofMillis(150000))
                .lifo()
                .build();
        HttpClient httpClient = HttpClient.create(connectionProvider)
                .metrics(true, Function.identity())
                .tcpConfiguration(tcpClient -> {
                    tcpClient.option(ChannelOption.TCP_NODELAY, true);
                    tcpClient.option(ChannelOption.SO_KEEPALIVE, true);
                    tcpClient.option(ChannelOption.ALLOW_HALF_CLOSURE, false);
                    return tcpClient;
                }).compress(true);

        final ExchangeStrategies strategies = ExchangeStrategies.builder()
            .codecs(codecs ->
                codecs.defaultCodecs().maxInMemorySize(
                    httpConnPoolProperties.getCodecMaxBytesToBuffer()))
            .build();

        return webClientBuilder
            .exchangeStrategies(strategies)
            .clientConnector(new ReactorClientHttpConnector(httpClient))
            .build();

AWS S3异步客户端配置 #

SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
            .connectionMaxIdleTime(Duration.ofMillis(75000))
            .tcpKeepAlive(true)
            .build();
        S3Configuration serviceConfiguration = S3Configuration.builder()
            .checksumValidationEnabled(false)
            .chunkedEncodingEnabled(false)
            .build();
        S3AsyncClientBuilder b = S3AsyncClient.builder()
            .httpClient(httpClient)
            .region(Region.of(cuePointBucketS3Region))
            .serviceConfiguration(serviceConfiguration);

        return b.build();

S3文件读取调度代码 #

downStreamService.get(tenantId)
                .publishOn(Schedulers.boundedElastic())
                .map(res -> {
                    return  parseResponse(res);
                })
                .map(data -> {
                    return cuePointFileLoader.loadAsync(data.getData());
                }).publishOn(Schedulers.boundedElastic())
                .flatMap(Mono::fromFuture)
                .map(BytesWrapper::asByteArray)
                .publishOn(Schedulers.boundedElastic())
                .subscribe(d -> {
                    try {
                        
                        CuePoints cuePoints = CuePoints.parseFrom(d);
                        CUE_POINT_ATOMIC_REF.lazySet(cuePoints);
                         metricClient.timer(Metric.CUE_POINT_S3_LOAD_TIME,
                            stopWatch.getTime(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
                    
                    } catch (Exception ex) {
                        return CuePoints.newBuilder().build();
                    } finally {
                        stopWatch.stop();
                    }
                });

1. Netty线程池资源竞争：WebClient和AWS S3异步客户端默认共享Netty的EventLoop线程池。S3文件读取操作（尤其是大文件）会占用大量EventLoop线程资源，导致WebClient调用下游服务时无法获取足够线程，引发延迟上升。
2. Schedulers.boundedElastic()滥用：调度代码中多次无意义地切换到boundedElastic线程池，过度切换会增加线程上下文切换开销；同时如果S3读取的阻塞操作（比如parseFrom）占用过多boundedElastic线程，也会影响其他依赖该线程池的操作。
3. S3客户端线程配置缺失：AWS S3异步客户端的Netty配置未指定独立的EventLoopGroup，默认复用全局线程池，加剧资源竞争。
4. 并行调用的连锁效应：下游服务调用是并行执行的，单个调用延迟上升会导致整体响应时间被最慢的调用拖长，形成连锁反应。

1. 为WebClient和S3客户端配置独立的Netty线程池 #

修改WebClient的HttpClient配置，指定独立EventLoopGroup #

// 创建独立的EventLoopGroup用于WebClient
EventLoopGroup webClientEventLoopGroup = new NioEventLoopGroup(
    Runtime.getRuntime().availableProcessors() * 2, // 根据实际并发需求调整线程数
    new DefaultThreadFactory("webclient-event-loop")
);

HttpClient httpClient = HttpClient.create(connectionProvider)
        .metrics(true, Function.identity())
        .tcpConfiguration(tcpClient -> {
            tcpClient.option(ChannelOption.TCP_NODELAY, true);
            tcpClient.option(ChannelOption.SO_KEEPALIVE, true);
            tcpClient.option(ChannelOption.ALLOW_HALF_CLOSURE, false);
            // 指定独立的EventLoopGroup
            return tcpClient.runOn(webClientEventLoopGroup);
        }).compress(true);

修改S3异步客户端配置，指定独立EventLoopGroup #

// 创建独立的EventLoopGroup用于S3客户端
EventLoopGroup s3EventLoopGroup = new NioEventLoopGroup(
    Runtime.getRuntime().availableProcessors(), // 根据S3读取频率和文件大小调整
    new DefaultThreadFactory("s3-event-loop")
);

SdkAsyncHttpClient httpClient = NettyNioAsyncHttpClient.builder()
        .connectionMaxIdleTime(Duration.ofMillis(75000))
        .tcpKeepAlive(true)
        .eventLoopGroup(s3EventLoopGroup) // 指定独立EventLoopGroup
        .build();

2. 优化调度代码的线程切换逻辑 #

去掉无意义的publishOn调用，仅在真正需要阻塞的操作（比如parseFrom）时切换到boundedElastic：

downStreamService.get(tenantId)
        .map(this::parseResponse) // 非阻塞操作，无需切换线程
        .map(data -> cuePointFileLoader.loadAsync(data.getData()))
        .flatMap(Mono::fromFuture) // S3异步调用，无需切换线程
        .map(BytesWrapper::asByteArray)
        .publishOn(Schedulers.boundedElastic()) // 仅在阻塞解析时切换
        .subscribe(d -> {
            stopWatch.stop();
            try {
                CuePoints cuePoints = CuePoints.parseFrom(d);
                CUE_POINT_ATOMIC_REF.lazySet(cuePoints);
                metricClient.timer(Metric.CUE_POINT_S3_LOAD_TIME,
                    stopWatch.getTime(TimeUnit.MILLISECONDS), TimeUnit.MILLISECONDS);
            } catch (Exception ex) {
                CUE_POINT_ATOMIC_REF.lazySet(CuePoints.newBuilder().build());
            }
        });

注意：如果parseResponse本身是阻塞操作，也需要将其放到publishOn之后执行。

3. 调整连接池和线程池参数 #

• WebClient连接池：根据下游服务的并发需求调整maxConnections，避免连接池过大导致资源浪费；同时合理设置pendingAcquireMaxCount，防止请求排队过长。
• S3客户端线程数：根据S3文件读取的频率和文件大小调整EventLoopGroup的线程数，避免线程过多导致上下文切换开销。
• boundedElastic线程池：通过配置spring.reactor.schedulers.bounded-elastic.max-size调整其最大线程数，确保阻塞操作有足够线程处理，同时避免线程泛滥。

4. 隔离S3调度任务的执行 #

将S3文件读取任务放在独立的调度线程池中，避免与WebClient的业务线程池竞争：

// 创建独立的调度线程池
ScheduledExecutorService s3Scheduler = Executors.newSingleThreadScheduledExecutor(
    new DefaultThreadFactory("s3-scheduler")
);

// 使用该线程池执行S3读取任务
s3Scheduler.scheduleAtFixedRate(() -> {
    // 放入S3读取逻辑
}, initialDelay, period, TimeUnit.MINUTES);

或者使用Spring的@Scheduled时指定自定义线程池：

@Configuration
public class SchedulerConfig {
    @Bean(name = "s3Scheduler")
    public TaskScheduler s3TaskScheduler() {
        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
        scheduler.setPoolSize(1);
        scheduler.setThreadNamePrefix("s3-scheduler-");
        scheduler.initialize();
        return scheduler;
    }
}

// 在调度方法上指定线程池
@Scheduled(fixedRate = 5, timeUnit = TimeUnit.MINUTES)
@Async("s3Scheduler")
public void loadCuePointsFromS3() {
    // S3读取逻辑
}

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