总阅读量:次

⭐⭐⭐ Spring Boot 项目实战 ⭐⭐⭐ Spring Cloud 项目实战
《Dubbo 实现原理与源码解析 —— 精品合集》 《Netty 实现原理与源码解析 —— 精品合集》
《Spring 实现原理与源码解析 —— 精品合集》 《MyBatis 实现原理与源码解析 —— 精品合集》
《Spring MVC 实现原理与源码解析 —— 精品合集》 《数据库实体设计合集》
《Spring Boot 实现原理与源码解析 —— 精品合集》 《Java 面试题 + Java 学习指南》

摘要: 原创出处 juejin.cn/post/7482765568123977764 「AAA_boy」欢迎转载，保留摘要，谢谢！

🙂🙂🙂关注微信公众号：【芋道源码】有福利：

  1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
  2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
  3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
  4. 新的源码解析文章实时收到通知。每周更新一篇左右。
  5. 认真的源码交流微信群。

在 Java 开发中，处理大规模数据时，常需要将列表拆分为多个子列表并异步处理，最后汇总结果。本文将介绍如何在 Spring Boot 中利用 Function<List<T>, ProcessResult> 和异步线程池实现这一需求，确保代码的高效性和可维护性。

一、实现思路

  1. 结果封装：定义 ProcessResult 类，封装每个子列表处理后的 int sum 和 StringBuilder msg。
  2. 线程池配置：创建独立的线程池，避免与其他异步任务干扰。
  3. 列表处理类：将列表按指定大小拆分，每个子列表通过 Function 处理并提交到线程池异步执行。
  4. 结果汇总：假设主线程需要获取到处理结果的成功记录数以及异常信息的描述，收集所有子任务的结果，计算总和并拼接消息，根据实际需求可以改成其他数据的收集。

二、代码实现

1. 结果封装类 ProcessResult

public class ProcessResult {
    privateint 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

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.concurrent.Executor;

@Configuration
publicclass 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

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
publicclass 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

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
publicclass 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(" ");
            }
            returnnew 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());
    }
}

核心流程：

  1. 初始化列表并定义拆分大小。
  2. 使用 Function 定义每个子列表的处理逻辑，计算 sum 并拼接 msg。
  3. 调用 ListProcessor 的 processList 方法异步处理所有子列表。
  4. 汇总所有子任务的结果，输出最终的总和与拼接后的消息。

三、代码解释

  • ProcessResult 类：封装每个子任务的处理结果，确保数据传递的清晰性。
  • 线程池配置：通过独立线程池避免资源竞争，提高异步处理的效率。
  • ListProcessor 类：负责列表拆分、异步任务提交和结果收集，利用 Spring 的 AOP 代理机制确保异步方法生效。
  • 主类逻辑：通过 Function 灵活定义业务逻辑，解耦数据处理与异步执行，使代码更具扩展性。

四、总结

本文介绍的方案通过 Function 和异步线程池实现了列表的高效拆分与处理，具有以下优点：

  1. 线程安全：使用独立线程池和 CompletableFuture 确保异步任务的安全执行。
  2. 代码解耦：将列表拆分、异步执行和业务逻辑分离，提高代码的可维护性。
  3. 结果清晰：通过 ProcessResult 封装结果，便于后续汇总和处理。

此方案适用于需要处理大规模数据并需要异步执行的场景，可根据实际需求调整线程池参数和业务逻辑，具有良好的灵活性和扩展性。

文章目录
  1. 1. 一、实现思路
  2. 2. 二、代码实现
    1. 2.1. 1. 结果封装类 ProcessResult
    2. 2.2. 2. 线程池配置类 ThreadPoolConfig
    3. 2.3. 3. 列表处理类 ListProcessor
    4. 2.4. 4. 主应用类 MainApplication
  3. 3. 三、代码解释
  4. 4. 四、总结