马克社区,马克吐温社区,马克java社区SpringBoot系列（16）：线程池-多线程Executors并发编程之批量查询-插入数据

SpringBoot系列（16）：线程池-多线程Executors并发编程之批量查询-插入数据

作者：修罗debug
版权声明：本文为博主原创文章，遵循 CC 4.0 by-sa 版权协议，转载请附上原文出处链接和本声明。

摘要：在上篇文章中Debug给大家分享介绍了“Java线程池-多线程的其中一种应用场景~广播式给所有有效用户发送邮件（通知）”，本篇文章我们将继续向前迈进，继续介绍并实战“线程池-多线程的应用场景”，这一场景简称为“批量插入大量的数据”，同样是采用Java中的Executors下的其中某种线程池进行实战实现！

内容：“批量插入数据”这一业务场景在企业级应用开发中还是比较常见的，顾明思议，即“将给定的大批量的数据插入到指定的数据库中去”，本文我们将采用Java代码、线程池-多线程的方式进行实现，感受一下开辟N个子线程去批量插入数据时的高效之处！

需要指出的是，“待插入的大批量的数据”的某个字段来源于数据库表“codes”，即字段item_id，在这一业务场景中我们将首先从该数据库表codes中拉取出所有的item_id，然后复制给另外的数据库表item_data中另外的字段code 中去，即实现所谓的“查询-批量插入大批量的数据”。

其中，目标数据库表item_data的DDL定义如下所示：

CREATE TABLE `item_data` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `code` varchar(255) CHARACTER SET utf8mb4 DEFAULT NULL COMMENT '编码',
  `p_id` int(11) DEFAULT NULL COMMENT '编号',
  `create_time` datetime DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8 COMMENT='数据表';

值得一提的是，那些待拉取的数据来源于数据库表codes中，在实战实现“插入大批量的数据”之前，我们写个Java单元测试或者其他的方式往codes表中模拟生成几十万甚至几百万数据记录，即codes数据库表中item_id的取值需要提前大批量生成并插入进去，在这里，我们预先生成的数据量为40w。

接下来，我们便进入实际的代码实战实现环节！

（1）首先，我们直接在ThreadController中写一个请求方法，待会儿用于Postman发起“批量插入数据”的请求，其代码如下所示：

    @RequestMapping(value = "all/insert/data",method = RequestMethod.GET)
    public BaseResponse insertAllData(){
        BaseResponse response=new BaseResponse(StatusCode.Success);
        try {
            threadService.insertDatas();

        }catch (Exception e){
            response=new BaseResponse(StatusCode.Fail.getCode(),e.getMessage());
        }
        return response;
    }

（2）threadService.insertDatas(); 即为批量插入大批量数据的实际核心业务逻辑，其完整的核心业务代码如下所示：

    @Autowired
    private DataMapper dataMapper;

    @Autowired
    private DataService dataService;

    //TODO:批量拉取、插入数据 - 40w
    public void insertDatas() throws Exception{
        //TODO:总线程数 10
        final Integer threadSize=10;

        //TODO:总数据量 40w
        final Long total=dataMapper.getTotal();

        if (total>0){
            //TODO:每个线程将执行插入操作的数据条目
            Long pageSize=(total%threadSize==0)?total/threadSize:total/threadSize+1;

            Set<String> datas;
            ExecutorService executorService=Executors.newFixedThreadPool(threadSize);

            List<ThreadInsertDataDto> list=Lists.newLinkedList();
            for (Long i=1L;i<=threadSize;i++){
                //TODO:将每个线程即将执行的具体条目记录拿出来
                if (Objects.equals(i, threadSize)){
                    pageSize = total - (threadSize-1) * pageSize;
                }
                datas=dataService.pageLimitData(i,pageSize);

                //TODO：构造线程实例
                list.add(new ThreadInsertDataDto(dataService,datas));
            }
            //TODO:多线程批量插入数据-逻辑
            executorService.invokeAll(list);
        }
    }

在该核心业务逻辑中，我们首先是开辟了10个线程（大伙儿要根据实际的机器配置此参数哈，我的机子是8核16线程的），根据这10个线程，按照“待拉取的总数据条目/10”得到的“数据量pageSize”即为每个线程要去数据库表codes中拉取的数据量！

每个线程拉取到相应的数据条目之后，即可构造对应的线程实例ThreadInsertDataDto，该类实例本质上就是一个“线程实例”，其中实现的run方法即为核心的“将数据插入到指定的数据表中”！

（3）ThreadInsertDataDto的完整代码如下所示：

public class ThreadInsertDataDto implements Callable<Boolean>{

    private DataService dataService;
    private Set<String> set;

    public ThreadInsertDataDto(DataService dataService, Set<String> set) {
        this.dataService = dataService;
        this.set = set;
    }

    //TODO:实际的插入数据到数据库表的真正逻辑
    @Override
    public Boolean call() throws Exception {
        if (dataService!=null){
            dataService.insertBatchData(set);
        }
        return true;
    }
}

而dataService.insertBatchData(set); 就是具体的“插入大批量的数据到数据库表的实际代码实现”，其完整的源代码如下所示：

    @Autowired
    private ItemDataMapper itemDataMapper;


    //TODO:批量插入数据
    @Async("taskDataExecutor")
    public void insertBatchData(Set<String> set){
        log.info("----开始批量插入数据----");

        List<ItemData> list= Lists.newLinkedList();
        //TODO:真正的插入数据的业务逻辑
        set.forEach(s -> {
            ItemData data=new ItemData(null,s,1);
            list.add(data);
        });
	    //这是一个批量插入的方法
        itemDataMapper.insertBatch(list);
    }

其中，itemDataMapper.insertBatch(list); 即为Mybatis批量插入的方法，其完整的源代码如下所示：

<!--批量插入-->
  <insert id="insertBatch">
    insert into item_data (id, code, p_id)
    values

    <foreach collection="datas" item="data" separator="," >
      (null,#{data.code},#{data.pId})
    </foreach>

  </insert>

至此，我们已经撸完“采用线程池-多线程的方式实现批量插入大批量数据”的业务场景，下面我们将整个项目跑起来，并采用Postman发起请求，你会惊讶的发现40w的数据，采用10个线程去批量插入时，5秒的时间都不到就可以完成了（其实是1s而已 哈哈）！

如下图所示：