devbox
pgrmghi
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

关于Mybatis批量插入使用JDBC原生batch批处理以及mybatis的ExecutorType.BATCH的批处理和Mybatis的动态sql语句批处理_mybaits executebatch

作者：pgrmghi | 2024-02-05 22:09:32

赞

踩

mybaits executebatch

近期处理数据，总结一下

项目背景：从Mongo中定时转移前第7天文本到mysql数据库且生成文件,有各种不同的定时任务，有批次消息、发送明细消息等等很多的数据需要转移，也有老系统和新系统数据的转移，mysql使用的分区表；架构使用的ssm，定时任务使用的xxl-job分布式任务调度

上面都是废话；

1、JDBC原生batch处理400万条数据

  1. /**
  2.      * @Description: 批量新增发送明细数据一次性插入1000条
  3.      * @Param:
  4.      * @return:
  5.      * @Author: fanghuaiming
  6.      * @Date:
  7.      */
  8.     public void savebatchDtlBatch(List<SmsSendHistoryDetail> smsSendHistoryDetails) {
  9.         LOG.info("Fuction:savebatchDtlBatch start insert into mysql batch 【 sms_msg_dtl 】");
  10.         Connection conn = DataSourceUtils.getConnection(dataSource1Config.testDataSource());
  11.         long before = System.currentTimeMillis();
  12.         int count = 0;
  13.         LOG.info("Fuction:savebatchDtlBatch Connection has opend");
  14.         PreparedStatement pstmt = null;
  15.         try {
  16.             pstmt = conn.prepareStatement("INSERT INTO sms_msg_dtl  (unique_id,username,mtype,mobile,batch_id,channel_id,location,words,sub_succ,sub_fail,rpt_succ,rpt_fail,uprice,cprice,submit_time,_id,mdstr,extend) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)");
  17.             conn.setAutoCommit(false);
  18.         } catch (SQLException e) {
  19.             e.printStackTrace();
  20.         }
  21.         for (int i = 0; i < smsSendHistoryDetails.size(); i++) {
  22.             try {
  23.                 SmsSendHistoryDetail smsSendHistoryDetail = smsSendHistoryDetails.get(i);
  24.                 pstmt.setLong(1, null == smsSendHistoryDetail.getUniqueId() ? 0 : smsSendHistoryDetail.getUniqueId());
  25.                 pstmt.setInt(2, null == smsSendHistoryDetail.getUserName() ? 0 : smsSendHistoryDetail.getUserName());
  26.                 pstmt.setInt(3, null == smsSendHistoryDetail.getMtype() ? 0 : smsSendHistoryDetail.getMtype());
  27.                 pstmt.setLong(4, null == smsSendHistoryDetail.getMobile() ? 0 : smsSendHistoryDetail.getMobile());
  28.                 pstmt.setLong(5, null == smsSendHistoryDetail.getBatchId() ? 0 : smsSendHistoryDetail.getBatchId());
  29.                 pstmt.setInt(6, null == smsSendHistoryDetail.getChannelId() ? 0 : smsSendHistoryDetail.getChannelId());
  30.                 pstmt.setString(7, null == smsSendHistoryDetail.getLocation() ? "" : smsSendHistoryDetail.getLocation());
  31.                 pstmt.setString(8, null == smsSendHistoryDetail.getWords() ? "" : smsSendHistoryDetail.getWords());
  32.                 pstmt.setInt(9, null == smsSendHistoryDetail.getSubSucc() ? 0 : smsSendHistoryDetail.getSubSucc());
  33.                 pstmt.setInt(10, null == smsSendHistoryDetail.getSubFail() ? 0 : smsSendHistoryDetail.getSubFail());
  34.                 pstmt.setInt(11, null == smsSendHistoryDetail.getRptSucc() ? 0 : smsSendHistoryDetail.getRptSucc());
  35.                 pstmt.setInt(12, null == smsSendHistoryDetail.getRptFail() ? 0 : smsSendHistoryDetail.getRptFail());
  36.                 pstmt.setLong(13, null == smsSendHistoryDetail.getuPrice() ? 0 : smsSendHistoryDetail.getuPrice());
  37.                 pstmt.setLong(14, null == smsSendHistoryDetail.getcPrice() ? 0 : smsSendHistoryDetail.getcPrice());
  38.                 pstmt.setLong(15, null == smsSendHistoryDetail.getSubmitTime() ? 0 : smsSendHistoryDetail.getSubmitTime());
  39.                 pstmt.setString(16, null == smsSendHistoryDetail.get_id() ? "" : smsSendHistoryDetail.get_id());
  40.                 pstmt.setString(17, null == smsSendHistoryDetail.getMdstr() ? "" : smsSendHistoryDetail.getMdstr());
  41.                 pstmt.setString(18, null == smsSendHistoryDetail.getExtend()  || "".equals(smsSendHistoryDetail.getExtend()) ? "" : smsSendHistoryDetail.getExtend());
  42.                 pstmt.addBatch();
  43.                 int a = i + 1;
  44.                 if (a % 1000 == 0) {
  45.                     try {
  46.                         pstmt.executeBatch();
  47.                     } catch (BatchUpdateException e) {
  48.                         int[] updateCounts = e.getUpdateCounts();
  49.                         for (int j = 0; j < updateCounts.length; j++) {
  50.                             if(updateCounts[j]<0) {
  51.                                 LOG.info("Fuction:savebatchDtlBatch mysql 【 sms_msg_dtl 】 mybe get some exception : {} And this data is : {}",e.getMessage(),smsSendHistoryDetails.get(count * 1000 +j));
  52.                             }
  53.                         }
  54.                     }
  55.                     count += 1;
  56.                     conn.commit();
  57.                     pstmt.clearBatch();
  58.                 }
  59.             } catch (
  60.                     SQLException e) {
  61.                 LOG.info("Fuction:savebatchDtlBatch mysql 【 sms_msg_dtl 】  batch insert get some Exception : {} ", e);
  62.                 e.printStackTrace();
  63.             }
  64.         }
  65.         try {
  66.             try {
  67.                 pstmt.executeBatch();
  68.             } catch (BatchUpdateException e) {
  69.                 int[] updateCounts = e.getUpdateCounts();
  70.                 for (int j = 0; j < updateCounts.length; j++) {
  71.                     if(updateCounts[j]<0) {
  72.                         LOG.info("Fuction:savebatchDtlBatch mysql 【 sms_msg_dtl 】 mybe get some exception : {} And this data is : {}",e.getMessage(),smsSendHistoryDetails.get(count * 1000 +j));
  73.                     }
  74.                 }
  75.             }
  76.             conn.commit();
  77.             pstmt.clearBatch();
  78.             count = 0;
  79.         } catch (SQLException e) {
  80.             e.printStackTrace();
  81.             LOG.info("Fuction:savebatchDtlBatch mysql 【 sms_msg_dtl 】  pstmt take executeBatch had SQLException :{}", e);
  82.         } finally {
  83.             try {
  84.                     pstmt.close();
  85.             } catch (SQLException e) {
  86.                 e.printStackTrace();
  87.                 LOG.info("Fuction:savebatchDtlBatch mysql 【 sms_msg_dtl 】  pstmt take close had Exception :{}", e);
  88.             }
  89.         }
  90.         long after = System.currentTimeMillis();
  91.         LOG.info("Fuction:savebatchDtlBatch  had done insert into mysql batch 【 sms_msg_dtl 】And Time-consuming : {} milliseconds", after - before);
  92.         DataSourceUtils.releaseConnection(conn, dataSource1Config.testDataSource());
  93.         LOG.info("Fuction:savebatchDtlBatch Connection has released");
  94.     }

2、Mybatis的Batch批处理

  1.     @Async
  2.     @Override
  3.     public void batchInsertBackUp(List<SmsRptDetail> smsSendMsgRpts) {
  4.         try {
  5.             long before = System.nanoTime();
  6.             //连接需要加上 allowMultiQueries=true
  7.             //进行mybaties的批次操作
  8.             SqlSession sqlSession = test1SqlSessionFactory.openSession(ExecutorType.BATCH,false);
  9.             //获取Mapper接口
  10.             SmsSendMsgRptBackUpMapper smsSendMsgRptBackUpMapper = sqlSession.getMapper(SmsSendMsgRptBackUpMapper.class);
  11.             //并行遍历循环新增(非commit操作)
  12.             smsSendMsgRpts.parallelStream().forEach(smsSendMsgRpt ->{
  13.                 smsSendMsgRptBackUpMapper.insertSmsSendMsgRpt(smsSendMsgRpt);
  14.             });
  15.             //进行批次提交
  16.             sqlSession.commit();
  17.             //刷新执行结果可以进行断言操作
  18.             List<BatchResult> batchResults = sqlSession.flushStatements();
  19.             //清除缓存
  20.             sqlSession.clearCache();
  21.             //关闭连接
  22.             sqlSession.close();
  23.             long after = System.nanoTime();
  24.             LOG.info("SmsSendMsgRptBackUpServiceImpl:batchInsertBackUp delete count:{} had done and total Millisecond use : {}",batchResults.size(),after - before);
  25.             batchResults.clear();
  26.         } catch (Exception e) {
  27.             e.printStackTrace();
  28.             LOG.info(e.getMessage());
  29.         }
  30.     }
  1. <insert id="insertSmsSendMsgRpt" parameterType="com.sioo.hysms.scheduled.pojo.SmsRptDetail">
  2.         insert into sms_rpt_dtl_backup
  3.         (rpt_time,username,
  4.         mobile,rptcode, unique_id,
  5.         batch_id,channel_id,extend,
  6.         resend_num,resend_count,
  7.         user_sub_time,channel_sub_time,
  8.         resend,uprice,cprice,msgid,
  9.         custom_id,source,create_time,update_time) VALUES
  10.         (#{rptTime},#{username},#{mobile},#{rptcode}, #{uniqueId},#{batchId},
  11.         #{channelId},#{extend},#{resendNum},#{resendCount},#{userSubTime},
  12.         #{channelSubTime},#{resend},#{uprice},#{cprice},#{msgid},#{customId},
  13.         #{source},#{createTime},#{updateTime})
  14.     </insert>

3、Mybatis动态sql语句

  1.     @Async
  2.     @Override
  3.     public void batchInsertBackUps(List<SmsRptDetail> smsSendMsgRpts) {
  4.         try {
  5.             long before = System.currentTimeMillis();
  6.             smsSendMsgRptBackUpMapper.batchInsertBackUp(smsSendMsgRpts);
  7.             long after = System.currentTimeMillis();
  8.             LOG.info("SmsSendMsgRptBackUpServiceImpl:batchInsertBackUps insert had done and total Millisecond use : {}",after - before);
  9.         } catch (Exception e) {
  10.             e.printStackTrace();
  11.             LOG.info(e.getMessage());
  12.         }
  13.     }
  1. <insert id="batchInsertBackUp" parameterType="java.util.List">
  2.         INSERT into sms_rpt_dtl_backup (rpt_time,username,
  3.         mobile,rptcode, unique_id,
  4.         batch_id,channel_id,extend,
  5.         resend_num,resend_count,
  6.         user_sub_time,channel_sub_time,
  7.         resend,uprice,cprice,msgid,
  8.         custom_id,source,create_time,update_time) VALUES
  9.         <foreach collection="smsSendMsgRpts" item="item" index="index" separator=",">
  10.             (#{item.rptTime,jdbcType=BIGINT},#{item.username,jdbcType=INTEGER},#{item.mobile,jdbcType=BIGINT},
  11.             #{item.rptcode,jdbcType=VARCHAR}, #{item.uniqueId,jdbcType=BIGINT},#{item.batchId,jdbcType=BIGINT},
  12.             #{item.channelId,jdbcType=INTEGER},#{item.extend,jdbcType=VARCHAR},#{item.resendNum,jdbcType=INTEGER},
  13.             #{item.resendCount,jdbcType=INTEGER},#{item.userSubTime,jdbcType=BIGINT},#{item.channelSubTime,jdbcType=BIGINT},
  14.             #{item.resend,jdbcType=INTEGER},#{item.uprice,jdbcType=BIGINT},#{item.cprice,jdbcType=BIGINT},
  15.             #{item.msgid,jdbcType=VARCHAR},#{item.customId,jdbcType=VARCHAR},#{item.source,jdbcType=INTEGER},
  16.             #{item.createTime,jdbcType=TIMESTAMP},#{item.updateTime,jdbcType=TIMESTAMP})
  17.         </foreach>
  18.  
  19.     </insert>

这些部分方法加了异步

  1. import org.springframework.beans.factory.annotation.Value;
  2. import org.springframework.context.annotation.Configuration;
  3. import org.springframework.context.annotation.PropertySource;
  4. import org.springframework.scheduling.annotation.AsyncConfigurer;
  5. import org.springframework.scheduling.annotation.EnableAsync;
  6. import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
  7.  
  8. import java.util.concurrent.Executor;
  9.  
  10. /**
  11. * @Description: 异步配置对象
  12. * @Param:
  13. * @return:
  14. * @Author: fanghuaiming
  15. * @Date:
  16. */
  17.  
  18. @Configuration
  19. @EnableAsync
  20. @PropertySource({"classpath:application-async.properties"})
  21. public class SpringAsyncConfig implements AsyncConfigurer {
  22.  
  23.     @Value("${springasync.corePoolSize}")
  24.     private int corePoolSize = Runtime.getRuntime().availableProcessors();
  25.     @Value("${springasync.maxPoolSize}")
  26.     private int maxPoolSize = Runtime.getRuntime().availableProcessors();
  27.  
  28.     /** 
  29.     * @Description: CPU密集型
  30.     * @Param:
  31.     * @return:
  32.     * @Author: fanghuaiming
  33.     * @Date:
  34.     */ 
  35.     @Override
  36.     public Executor getAsyncExecutor() {
  37.         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
  38.         executor.setCorePoolSize(corePoolSize);
  39.         executor.setMaxPoolSize(maxPoolSize);
  40.         executor.setThreadNamePrefix("HysmsScheduled-");
  41.         executor.initialize();
  42.         return executor;
  43.     }
  44. }

效率总结：

1、实际上从每日4000万的数据来看，速度最快的反而是JDBC这种操作，但是如果插入过程中出现什么问题会造成数据库连接卡在这里导致数据库连接不够，因为始终没有归还，从第一个jdbc原生可以看出来我只是捕获了异常而已，也只是业务上的处理，JDBC的连接如果处理最终也只是close掉而不是归还给某个池化的连接，但是不可否认它的速度够快

2、其实我打心眼里是想大规模使用Mybatis的Executor的Batch，它是Mybatis提供的方便简洁美观，但是实际上它的效率很慢，曾经接触过部分了解，这个Batch实际上是有大小的限制的，例如1万条过大的sql拼接到一个类似buffer，这个buffer是有大小限制的，需要你进行清除或者按照第一种JDBC操作的方式达到多少阀值进行一次提交的操作，会有一些坑，这里我只是贴出来保证可以执行，至于效率我并没有优化，可能需要自己进行优化，我因为任务加身就没有继续做了，有兴趣的可以多看看资料或者《深入浅出MyBatis》这个pdf我会传到我的下载里，不知道怎么设置不需要积分，不想花积分可以进我们的讨论群，稍后贴出

3、Mybatis的动态sql语句实际上官方是推崇的，第三种方式我是进行了动态规划做的子任务(使用Fork-Join也是一样的)，这个方法只进行300万的数据的插入，我仔细观察过效率(广义上的观察，也就是人工观察?)，在前100万条的插入的时候是非常快的，但是后面的300万条数据实际还是很慢的，感觉和Executor的BATCH效率差不多，翻了官方的Mybatis文档也没有介绍详细的效率问题，更多的是如何使用，当然这也可能和它的动态sql拦截器相关执行有关，我没有进行优化了，因为现在任务正常跑，暂时不需要进行优化，还是推荐大家多看看《深入浅出MyBatis》

统一说明：

这里使用的是mysql，如果使用Oracle其实也差不多，改一改一样的，我的mysql使用的是InnoDB数据库引擎，最近很多博客标题都是教你如何3分钟插入1000万条数据都是扯淡，你翻进去一看是MyISAM，那连键锁、临键锁、间隙锁都没有当然快了，为了博噱头什么都不管不顾

还有，我说的4000万条数据中业务会做很多事情，例如转换mongo的数据的判断、聚合、条件的整合还有生成csv文件消耗的大量IO等等业务上的操作，所以我是人工查看的效率而不是代码上进行监控，如果有想做这个监控的可以自己写自定义注解加在方法头上进行AOP切面监控，用的地方用，不用的地方就不加去进行监控

总结：

总的来说我个人比较推崇第一种方法(除了手动关闭连接之外)，原生的有时候比较方便，第三方的就需要阅读人家的源码去进行优化了，在以上代码部分有疑问或需要下载pdf的请进讨论：java华夏一区，禁止推销

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/pgrmghi/article/detail/61854
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号