当前位置：首页 > news >正文

SpringBoot中使用多线程ThreadPoolTaskExecutor+CompletableFuture

news 2024/11/2 13:41:20

SpringBoot中使用多线程ThreadPoolTaskExecutor+CompletableFuture

定义一个线程池，并将其注入为bean

我使用的是spring提供的线程池，所以不需要写关闭的逻辑

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;import java.util.concurrent.Executor;@Configuration
@EnableAsync
public class ExecutorConfig {@Bean(name = {"saveExecutor"})public Executor taskExecutor() {ThreadPoolTaskExecutor executor =  new ThreadPoolTaskExecutor();executor.setCorePoolSize(5);//核心线程数executor.setMaxPoolSize(1000);// 最大线程数executor.setQueueCapacity(500);// 任务队列容量executor.setThreadNamePrefix("taskExecutor-");executor.initialize(); //初始化线程池return executor;}
}

使用

import cn.hutool.core.collection.ListUtil;
import com.train.demo.entities.User;
import com.train.demo.repository.UserRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;@Service
@Slf4j
public class UserService {private final UserRepository userRepository;private final Executor saveExecutor;@Autowiredpublic UserService(UserRepository userRepository,// 依赖注入，一定要使用@Qualifier指定bean名称，不然有问题@Qualifier("saveExecutor") Executor saveExecutor) {this.userRepository = userRepository;this.saveExecutor = saveExecutor;}public boolean addAllUsers() {// 定义一个数组，并生成1000000条数据ArrayList<User> users = new ArrayList<User>(){{for (int i = 0; i < 1000000; i++) {add(new User(null,"user"+i,2,"男"));}}};int count = 0;// 定义一个接收“CompletableFuture返回”的listList<CompletableFuture<Integer>> futureList = new ArrayList<>();while( count < users.size() ){// 这里只是计算剩下的数据量够不够1000，结果作为截取list的参数int size = users.subList( count, users.size()-1).size() > 9999 ? 9999: users.subList( count, users.size()-1).size();// 调用异步方法CompletableFuture<Integer> future = addAllUsersImpl(users.subList(count, count += size));futureList.add(future);count++;}futureList.forEach(CompletableFuture::join);//阻塞所有的异步方法全执行完毕再向下执行return true;}@Asyncprotected CompletableFuture<Integer> addAllUsersImpl(List<User> users) {// 此处使用spring线程池作为CompletableFuture线程池参数，并且异步执行return CompletableFuture.supplyAsync(() -> {log.info("save users thread[{}]", Thread.currentThread().getName());List<User> list = ListUtil.list(false, userRepository.saveAll(users));// 存储到数据库return list.size();// 返回存储的数据量}, saveExecutor);}
}