import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
@Service
public class MultiThreadedBatchService {
@Autowired
private YourRepository yourRepository;
@Transactional
public void batchInsert(List<YourEntity> entities) {
if (entities.isEmpty()) {
return;
}
int threads = Runtime.getRuntime().availableProcessors(); // 使用CPU核心数量的线程
int batchSize = (int) Math.ceil((double) entities.size() / threads);
ExecutorService executorService = Executors.newFixedThreadPool(threads);
List<List<YourEntity>> partitions = IntStream.range(0, threads)
.mapToObj(i -> entities.stream()
.skip(i * batchSize)
.limit(batchSize)
.collect(Collectors.toList()))
.collect(Collectors.toList());
partitions.forEach(partition -> executorService.submit(() -> {
yourRepository.saveAll(partition); // 假设saveAll方法支持批量插入
}));
executorService.shutdown(); // 确保所有任务执行完毕
try {
executorService.awaitTermination(10, TimeUnit.MINUTES); // 等待所有任务完成
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Multi-threaded batch insert failed to complete within timeout", e);
}
}
}这个代码示例展示了如何在Spring Boot应用中使用多线程批量插入数据。首先，它计算了需要的线程数和每个批次的大小。然后，它创建了一个固定大小的线程池，并将整个实体列表分成多个批次，每个线程处理一个批次。每个批次通过yourRepository.saveAll(partition)进行插入。最后，关闭线程池并等待所有任务完成。这个过程是在事务的上下文中进行的，以确保要么所有的批次都成功处理，要么全部回滚。