如何配置Spring Batch消费非阻塞HTTP资源？解决线程阻塞问题

问题本质分析 #

你遇到的情况非常典型：在8核客户端上，长期运行的全并行Spring Batch作业因为阻塞式HTTP请求占满了线程池，导致所有线程都卡在等待HTTP响应的状态——虽然客户端和服务器的CPU、内存都没跑满，但线程资源被耗尽，整个系统陷入停滞。这时候切换到非阻塞HTTP请求，就能让线程在等待响应时去处理其他任务，大幅提升线程利用率。

具体解决方案步骤 #

1. 引入Jetty9异步HTTP客户端依赖

首先在你的项目依赖中加入Jetty的异步HttpClient相关包（以Maven为例）：

<dependency>
    <groupId>org.eclipse.jetty</groupId>
    <artifactId>jetty-client</artifactId>
    <version>9.4.51.v20230217</version> <!-- 替换为你需要的稳定版本 -->
</dependency>

2. 配置Jetty异步HttpClient

创建一个配置类，初始化Jetty的异步HttpClient，开启非阻塞模式：

import org.eclipse.jetty.client.HttpClient;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class JettyHttpClientConfig {

    @Bean(destroyMethod = "stop")
    public HttpClient jettyHttpClient() throws Exception {
        HttpClient httpClient = new HttpClient();
        // 配置非阻塞相关参数，比如连接超时、请求超时
        httpClient.setConnectTimeout(5000);
        httpClient.setIdleTimeout(30000);
        // 启动客户端
        httpClient.start();
        return httpClient;
    }
}

3. 实现非阻塞的Spring Batch ItemReader

修改你的ItemReader，用Jetty的异步HttpClient发起请求，避免阻塞线程。这里可以用CompletableFuture来处理异步响应，让作业线程不会被绑定在等待HTTP响应上：

import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.http.HttpMethod;
import org.springframework.batch.item.ItemReader;
import org.springframework.stereotype.Component;

import java.util.concurrent.CompletableFuture;

@Component
public class NonBlockingHttpItemReader implements ItemReader<String> {

    private final HttpClient jettyHttpClient;
    private final String resourceUrl; // 你的目标HTTP资源地址

    public NonBlockingHttpItemReader(HttpClient jettyHttpClient) {
        this.jettyHttpClient = jettyHttpClient;
        this.resourceUrl = "http://your-server/api/data";
    }

    @Override
    public String read() throws Exception {
        // 发起异步HTTP请求，当前线程不会被阻塞
        CompletableFuture<ContentResponse> future = jettyHttpClient.newRequest(resourceUrl)
                .method(HttpMethod.GET)
                .sendAsync();

        // 等待响应（此时Jetty会用专属I/O线程处理响应回调，作业线程只是短暂等待）
        ContentResponse response = future.get();
        if (response.getStatus() == 200) {
            return response.getContentAsString();
        }
        // 返回null表示没有更多数据，或者根据业务处理异常场景
        return null;
    }
}

4. 调整Spring Batch的线程池配置

因为非阻塞请求不会长时间占用作业线程，你可以适当调整线程池参数，让更多并行任务同时运行，同时避免线程过多导致资源浪费：

import org.springframework.batch.core.launch.support.TaskExecutorJobLauncher;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.task.TaskExecutor;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

@Configuration
public class BatchConfig {

    @Bean
    public TaskExecutor batchTaskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(16); // 可根据实际情况调整，比如核数的2倍
        executor.setMaxPoolSize(32);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("batch-thread-");
        executor.initialize();
        return executor;
    }

    @Bean
    public TaskExecutorJobLauncher jobLauncher(JobRepository jobRepository, TaskExecutor batchTaskExecutor) throws Exception {
        TaskExecutorJobLauncher launcher = new TaskExecutorJobLauncher();
        launcher.setJobRepository(jobRepository);
        launcher.setTaskExecutor(batchTaskExecutor);
        launcher.afterPropertiesSet();
        return launcher;
    }
}

5. 验证效果

部署后观察客户端的线程状态：原来的阻塞线程会大幅减少，Jetty的I/O线程会处理HTTP响应的回调，作业线程可以快速释放去处理下一个任务。同时监控CPU和内存使用率，会发现资源利用率更合理，作业吞吐量明显提升。

注意事项 #

• 确保你的HTTP资源服务器不会在服务端长时间阻塞请求（即使服务端是阻塞的，客户端的非阻塞模式依然能提升线程利用率）
• 务必处理异步请求的异常情况，比如超时、连接失败，避免未捕获的异常导致作业中断
• 如果需要批量获取HTTP资源，可以考虑用Jetty的批量异步请求API，进一步提升处理效率

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