利用线程池优化大批量数据库操作

多线程对数据库进行操作时，并非线程数越多操作越快。

卡布奇诺-海晨

713人浏览 · 2024-04-09 11:15:19

卡布奇诺-海晨 · 2024-04-09 11:15:19 发布

博主介绍：✌全网粉丝5W+，全栈开发工程师，从事多年软件开发，在大厂呆过。持有软件中级、六级等证书。可提供微服务项目搭建与毕业项目实战，博主也曾写过优秀论文，查重率极低，在这方面有丰富的经验✌

博主作品：《Java项目案例》主要基于SpringBoot+MyBatis/MyBatis-plus+MySQL+Vue等前后端分离项目，可以在左边的分类专栏找到更多项目。《Uniapp项目案例》有几个有uniapp教程，企业实战开发。《微服务实战》专栏是本人的实战经验总结，《Spring家族及微服务系列》专注Spring、SpringMVC、SpringBoot、SpringCloud系列、Nacos等源码解读、热门面试题、架构设计等。除此之外还有不少文章等你来细细品味，更多惊喜等着你哦

🍅uniapp微信小程序🍅面试题软考题免费使用，还可以使用微信支付，扫码加群。由于维护成本问题得不到解决，可能将停止线上维护。

点击这里预览

抖音体验版

🍅文末获取联系🍅精彩专栏推荐订阅👇🏻👇🏻 不然下次找不到哟

Java项目案例《100套》
https://blog.csdn.net/qq_57756904/category_12173599.html
uniapp小程序《100套》

https://blog.csdn.net/qq_57756904/category_12173599.html

有需求代码永远写不完，而方法才是破解之道，抖音有实战视频课程，某马某千等培训都是2万左右，甚至广东有本科院校单单一年就得3万4年就12万学费，而且还没有包括吃饭的钱。所以很划算了。另外博客左侧有源码阅读专栏，对于求职有很大帮助，当然对于工作也是有指导意义等。在大城市求职，你面试来回一趟多多少少都在12块左右，而且一般不会一次性就通过，还得面试几家。而如果你对源码以及微服务等有深度认识，这无疑给你的面试添砖加瓦更上一层楼。

最后再送一句：最好是学会了，而不是学废了！！

在数据库中如果需要对大量的数据进行批量修改，并不是一项简单的工作。而利用线程池可以帮助我们解决这些问题

1.循环操作的代码

For循环插入5000条记录

@Test
void contextLoads6() {
    ArrayList<UserInfo> allUser = new ArrayList<>(5000);
 
    //生成size个UserInfo
    for (int i = 0; i < 5000; i++) {
        UserInfo userInfo = new UserInfo();
        userInfo.setId(String.valueOf(i + 10000));
        allUser.add(userInfo);
    }
 
    long  startTime = System.currentTimeMillis();
    for (int i = 0;i<5000;i++){
        mapper.insert(allUser.get(i));
    }
    System.out.println("单线程for循环插入5000耗时:"+(System.currentTimeMillis()-startTime));
}

耗时:14944ms

2.利用多线程

实现Callable接口，封装插入数据Task

/**
 * 插入SignUser数据任务，带回调结果
 * -
 * @author CeaM
 * 2024/4/9
 */
public class InsertTarget implements Callable<Integer> {

    private SignUserMapper mapper;

    private List<SignUser> list;

    private CountDownLatch countDownLatch;

    public InsertTarget(SignUserMapper mapper,
                        List<SignUser> list,
                        CountDownLatch countDownLatch) {
        this.mapper = mapper;
        this.list = list;
        this.countDownLatch = countDownLatch;
    }

    @Override
    public Integer call() throws Exception {
        // 记录插入成功的条数
        int rows = 0;
        for (SignUser user : list) {
            mapper.insert(user);
            rows++;
        }

        //完全插入成功，CountDownLatch-1
        if (rows == list.size()) {
            countDownLatch.countDown();
        }

        return rows;
    }
}

Service层:

@Slf4j
@Service
public class SignUserServiceImpl
        extends ServiceImpl<SignUserMapper, SignUser> implements SignUserService {

    @Override
    public int addSignUser(Integer size) {
        // 制造数据，list里面有size(5000)个UserInfo
        List<SignUser> allUser = this.createSizeList(size);
        // 线程数
        int threadSize = 5;

        ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(threadSize,
                threadSize, 10, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
        // 每个线程处理的数据量
        final Integer dataPartionLength = (allUser.size() + (threadSize) - 1) / (threadSize);
        log.info("每个线程消费：{}", dataPartionLength);

        int downLatchSize = 0;
        // 将downLatchSize这个值计算出来
        if (size % dataPartionLength == 0){
            downLatchSize = size / dataPartionLength;
        }else {
            downLatchSize = size / dataPartionLength + 1;
        }

        CountDownLatch latch = new CountDownLatch(downLatchSize);
        long startTime = System.currentTimeMillis();
        for (int i = 0; i < downLatchSize; i++) {
            // 每个线程均分需要处理的数据
            List<SignUser> threadDatas = allUser.stream()
                    .skip(i * dataPartionLength)
                    .limit(dataPartionLength).collect(Collectors.toList());

            // 调用实现了Callable的InsertTarget类，并将mapper，threadDatas，latch传进去
            InsertTarget insertTarget = new InsertTarget(baseMapper, threadDatas, latch);
            FutureTask<Integer> futureTask = new FutureTask<>(insertTarget);

            // 执行任务
            threadPoolExecutor.execute(futureTask);
        }
        try {
            // 保证每个线程完成后再完成计时
            latch.await();
            long endTime = System.currentTimeMillis();
            log.info("线程数为{}的线程池插入5000用时：{}", threadSize, endTime - startTime);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return 200;
    }

    /**
     * 用来生成数据，返回一个size大小的list
     */
    private List<SignUser> createSizeList(Integer size) {
        List<SignUser> list = new ArrayList<>(100);

        // 生成size个UserInfo，并放进List
        for (int i = 0; i < size; i++) {
            SignUser userInfo = new SignUser();
            userInfo.setId(String.valueOf(i + 1));
            userInfo.setCrtTime(new Date());
            list.add(userInfo);
        }
        return list;
    }
}

controller：

@RestController
@RequestMapping("/user")
public class SignUserController {

    private final SignUserService signUserService;

    public SignUserController(SignUserService signUserService) {
        this.signUserService = signUserService;
    }

    @GetMapping
    public void batchWithPool(@RequestParam Integer size) {
        signUserService.addSignUser(size);
    }
}