在 Java 开发中，处理大规模数据时，常需要将列表拆分为多个子列表并异步处理，最后汇总结果。本文将介绍如何在 Spring Boot 中利用 Function<List<T>, ProcessResult> 和异步线程池实现这一需求，确保代码的高效性和可维护性。
一、实现思路
- 结果封装：定义
ProcessResult类，封装每个子列表处理后的int sum和StringBuilder msg。 - 线程池配置：创建独立的线程池，避免与其他异步任务干扰。
- 列表处理类：将列表按指定大小拆分，每个子列表通过
Function处理并提交到线程池异步执行。 - 结果汇总：假设主线程需要获取到处理结果的成功记录数以及异常信息的描述，收集所有子任务的结果，计算总和并拼接消息，根据实际需求可以改成其他数据的收集。
二、代码实现
1. 结果封装类 ProcessResult
java
体验AI代码助手
代码解读
复制代码
public class ProcessResult {
private int sum;
private StringBuilder msg;
public ProcessResult(int sum, StringBuilder msg) {
this.sum = sum;
this.msg = msg;
}
public int getSum() {
return sum;
}
public StringBuilder getMsg() {
return msg;
}
}
作用：封装每个子列表处理的结果，便于后续汇总。
2. 线程池配置类 ThreadPoolConfig
java
体验AI代码助手
代码解读
复制代码
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;
@Configuration
public class ThreadPoolConfig {
@Bean(name ="customThreadPool")
public Executor customThreadPool() {
ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
executor.setCorePoolSize(5);
executor.setMaxPoolSize(10);
executor.setQueueCapacity(25);
executor.setThreadNamePrefix("custom-thread-");
executor.initialize();
return executor;
}
}
作用：配置独立的线程池 customThreadPool，可根据需求调整线程池参数（如核心线程数、最大线程数等）。建议：通过配置类和配置文件维护线程池参数，这里写成固定值
3. 列表处理类 ListProcessor
java
体验AI代码助手
代码解读
复制代码
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.function.Function;
@Component
public class ListProcessor<T> {
@Autowired
private ApplicationContext applicationContext;
@Autowired
private ThreadPoolTaskExecutor customThreadPool;
/**
* 异步处理主方法，负责拆分list以及异步调用处理方法，处理方法会返回数据
*/
public List<ProcessResult> processList(List<T> list, int chunkSize, Function<List<T>, ProcessResult> function)
throws InterruptedException, ExecutionException {
List<List<T>> chunks = splitList(list, chunkSize);
List<CompletableFuture<ProcessResult>> futures = new ArrayList<>();
ListProcessor<T> self = applicationContext.getBean(this.getClass());
for (List<T> chunk : chunks) {
futures.add(self.processChunkAsync(chunk, function));
}
List<ProcessResult> results = new ArrayList<>();
for (CompletableFuture<ProcessResult> future : futures) {
results.add(future.get());
}
return results;
}
/**
* 异步处理执行方法，通过Async注解实现异步效果
*/
@Async("customThreadPool")
public CompletableFuture<ProcessResult> processChunkAsync(List<T> chunk, Function<List<T>, ProcessResult> function) {
return CompletableFuture.completedFuture(function.apply(chunk));
}
/**
* 异步处理主方法2，负责拆分list以及异步调用处理方法，使用CountDownLatch保持主线程等待,无需处理方法返回数据时使用该方法，
* 使用这种方法，仍需要主线程获取处理结果的，可以在主线程中创建一个线程安全的对象，
* 在子线程的处理方法consumer中去操作该对象，例如使用Collections.synchronizedList创建的集合
*/
public <T> void processList2(List<T> list, int chunkSize, Consumer<List<T>> consumer) throws InterruptedException {
//list大小小于子集合大小，拆分只会拆出一个子集合，异步没有意义，直接同步执行
if (list.size() <= chunkSize) {
consumer.accept(list);
return;
} //按chunkSize拆分list为多个子集合
List<List<T>> chunks = splitList(list, chunkSize);
//创建一个CountDownLatch，方便主进程等待
CountDownLatch latch = new CountDownLatch(chunks.size());
for (List<T> chunk : chunks) {
//遍历子集合的集合，依次异步调用处理方法，通过bean调用实现异步
getSelf().processChunkAsync(chunk, consumer, latch);
} //阻塞主线程直到异步完成
latch.await();
}
/**
* 无需返回值的Consumer调用时的异步处理执行方法，通过Async注解实现异步效果
*/
@Async("customThreadPool")
public <T> void processChunkAsync(List<T> chunk, Consumer<List<T>> consumer, CountDownLatch latch) {
try {
consumer.accept(chunk);
} finally {
latch.countDown();
}
}
/**
* 拆分list为多个子集合，针对最后一个子集合可能存在下标越界的情况，需取集合剩余记录数和子集合大小的较小值
*/
private List<List<T>> splitList(List<T> list, int chunkSize) {
List<List<T>> chunks = new ArrayList<>();
for (int i = 0; i < list.size(); i += chunkSize) {
chunks.add(list.subList(i, Math.min(i + chunkSize, list.size())));
}
return chunks;
}
}
关键逻辑：
processList方法：拆分列表并提交异步任务，通过ApplicationContext获取当前类的 Bean 实例，确保异步注解生效。splitList方法：按指定大小将列表拆分为子列表。processChunkAsync方法：使用@Async标记异步执行，调用传入的Function处理子列表并返回结果。
4. 主应用类 MainApplication
java
体验AI代码助手
代码解读
复制代码
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableAsync;
import java.util.ArrayList;
import java.util.List;
import java.util.StringJoiner;
import java.util.function.Function;
@SpringBootApplication
@EnableAsync
public class MainApplication implements CommandLineRunner {
@Autowired
private ListProcessor<Integer> listProcessor;
public static void main(String[] args) {
SpringApplication.run(MainApplication.class, args);
}
@Override
public void run(String... args) throws Exception {
List<Integer> list = new ArrayList<>();
for (int i = 1; i <= 10; i++) {
list.add(i);
}
int chunkSize = 3;
// 定义子列表的处理逻辑
Function<List<Integer>, ProcessResult> function = chunk -> {
int sum = 0;
StringBuilder msg = new StringBuilder();
for (int num : chunk) {
sum += num;
msg.append(num).append(" ");
}
return new ProcessResult(sum, msg);
};
// 执行列表处理并获取结果
List<ProcessResult> results = listProcessor.processList(list, chunkSize, function);
// 汇总结果
int totalSum = 0;
StringJoiner joiner = new StringJoiner(", ");
for (ProcessResult result : results) {
totalSum += result.getSum();
joiner.add(result.getMsg().toString());
}
System.out.println("sum= " + totalSum);
System.out.println("msg= " + joiner.toString());
}
}
核心流程：
- 初始化列表并定义拆分大小。
- 使用
Function定义每个子列表的处理逻辑，计算sum并拼接msg。 - 调用
ListProcessor的processList方法异步处理所有子列表。 - 汇总所有子任务的结果，输出最终的总和与拼接后的消息。
三、代码解释
ProcessResult类：封装每个子任务的处理结果，确保数据传递的清晰性。- 线程池配置：通过独立线程池避免资源竞争，提高异步处理的效率。
ListProcessor类：负责列表拆分、异步任务提交和结果收集，利用 Spring 的 AOP 代理机制确保异步方法生效。- 主类逻辑：通过
Function灵活定义业务逻辑，解耦数据处理与异步执行，使代码更具扩展性。
四、总结
本文介绍的方案通过 Function 和异步线程池实现了列表的高效拆分与处理，具有以下优点：
- 线程安全：使用独立线程池和
CompletableFuture确保异步任务的安全执行。 - 代码解耦：将列表拆分、异步执行和业务逻辑分离，提高代码的可维护性。
- 结果清晰：通过
ProcessResult封装结果，便于后续汇总和处理。
此方案适用于需要处理大规模数据并需要异步执行的场景，可根据实际需求调整线程池参数和业务逻辑，具有良好的灵活性和扩展性。
