[大数据]如何快速插入10万条数据?

1、目的及解决思路

目的：在Mysql中，插入10w条测试数据

解决思路：利用Mybatis-plus的手动批量插入+多线程来插入数据

2、具体实现

• 线程池的创建

  /**
   * @author: tianjx
   * @date: 2021/12/12 17:19
   * @description: 线程池工具类
   */
  public class ThreadPoolUtil {
      /**
       * 核心线程数量：8
       * 最大线程数：16（CPU密集型，一般是电脑核数 * 2）
       * 多余的空闲线程存活时间：30
       * 存活时间单位：秒
       * 阻塞队列：LinkedBlockingDeque
       * 拒绝策略：CallerRunsPolicy
       *
       */
      public static final ThreadPoolExecutor THREAD_POOL = new ThreadPoolExecutor(8,16,
              30, TimeUnit.SECONDS, new LinkedBlockingDeque<>(), new ThreadPoolExecutor.CallerRunsPolicy());
  }
  

• 批量插入语句（Dao层，重点：批量插入语句）

  /**
   * @author: tianjx
   * @date: 2021/12/12 17:00
   * @description: 批量插入Dao
   */
  @Repository
  public interface TmpDataOperationDao{
  
      /**
       * 插入需求部门数据
       * @return
       */
      @Insert("<script>" +
              "INSERT INTO l_recorder_org_name ( id, recorderOrgName, projectId ) VALUES" +
              "<foreach collection='recorders' item='recorder'   separator=','> " +
              "(replace(UUID(),'-',''),#{recorder.recorderOrgName},#{recorder.projectId})" +
              "</foreach> " +
              "</script>")
      Boolean insRecorderOrgName(@Param("recorders") List<HashMap<String,String>> recorders);
  }
  

• service层（普通调用）

  /**
   * @author: tianjx
   * @date: 2021/12/12 17:00
   * @description: 服务层
   */
  public interface TmpDataOperationService{
      /**
       * 插入需求部门数据
       * @return
       */
      Boolean insRecorderOrgName();
  }
  

  /**
   * @author: tianjx
   * @date: 2021/12/12 17:05
   * @description: 服务实现层
   */
  @Slf4j
  @Service
  public class TmpDataOperationServiceImpl implements TmpDataOperationService {
  
      @Autowired
      TmpDataOperationDao tmpDataOperationDao;
  
      /**
       * 插入需求部门数据，中标公司数据
       *
       * @return
       */
      @Override
      public Boolean insRecorderOrgName() {
          try {
              List<HashMap<String,String>> lRecorderOrgNameList = new ArrayList<>();
              for (int j = 0; j < 5000; j++) {
                  HashMap<String,String> lRecorderOrgName = new HashMap<>();
                  lRecorderOrgName.put("recorderOrgName","test");
                  lRecorderOrgName.put("projectId","1");
                  lRecorderOrgNameList.add(lRecorderOrgName);
              }
              tmpDataOperationDao.insRecorderOrgName(lRecorderOrgNameList);
          }catch (Exception e){
              e.printStackTrace();
          }
          return true;
      }
  }
  

• controller层（重点：线程池）

  /**
   * @author: tianjx
   * @date: 2021/12/12 17:14
   * @description:
   */
  @RestController
  @CrossOrigin
  @Slf4j
  @RequestMapping("/tmpData")
  public class TmpDataOperationController {
  
      @Autowired
      private TmpDataOperationService tmpDataOperationService;
  
  
      @GetMapping("/insRecorderOrgName")
      @ResponseBody
      public Response getTmpJsonData() throws InterruptedException {
          long startTimeMillis = System.currentTimeMillis();
          int num = 20;
          CountDownLatch countDownLatch = new CountDownLatch(num);
          for (int i = 0; i < num; i++) {
              ThreadPoolUtil.THREAD_POOL.execute(
                      () -> {
                          tmpDataOperationService.insRecorderOrgName();
                          countDownLatch.countDown();// 线程计数器，递减
                      }
              );
          }
  
          countDownLatch.await();// 等带其他线程结束，在执行主线程
          long endTimeMillis = System.currentTimeMillis();
          System.out.println("线程" + Thread.currentThread().getId() + "执行结束，用时：" + (endTimeMillis - startTimeMillis) + "ms");
  
          return new Response<>().success();
      }
  }