使用ibatis处理大数据量批量插入更新问题

时间:2022-09-22 00:06:26

近期项目中遇到一个问题,有一批数据,少则几百条,多则上万条,需要向数据库中进行插入和更新操作,即:当数据库中有数据时,更新之,没有数据时,插入之。

解决问题的步骤如下:

首先想到的当然是用先根据条件select count(*) from table where "case",判断select出的结果,如果结果>0则更新,等于0则插入。

 public Class intefaceImpl implements XXXService{
     //.......
     
     public Result insertOrUpdateXXDOBatch(List<XXDO> xxDos) throws DAOException{
         for(XXDO xxDo:xxDos){
             //IbatisDAO.queryTotalCount对应select count(*)语句
             //  xxDo.getCase()代表的是要更新的条件
             int count=(int)sqlMapClientTemplate.queryForObject("IbatisDAO.queryTotalCount", xxDo.getCase());
             if(count==0){
 sqlMapClientTemplate.update("IbatisDAO.update", xxDo);
         }else {
             sqlMapClientTemplate.insert("IbatisDAO.insert", xxDo);
         }   
   }
 } 

随后发现这个有漏洞,这些漏洞也是这个问题的难点:

1,数据量大,容易引起大量的数据库读写操作,对数据库压力大,而且在读写过程中通信消耗也大;

2,由于读写时整体放到一个服务接口中的,所以服务会超时。

现在需要一步步解决这些问题,如何解决呢?首先,对于数据量大造成的压力大和通信消耗大的问题,可以想到的一种方法,就是批量执行,恰好,ibatis就提供了批量执行的方法startBatch/endBatch。

另外还要思考能否简化数据库的操作呢?主要的方法,就是不再使用select,方法有两种:

1,因为我用的是mysql数据库,所以,可以使用replace语句或on duplicate key update来做(具体语句写法这里就不多写了,随便一搜都能查到);

2,先update一次,如果update的返回值为0,表示update了0行,则insert。

首先我考察了第一种方法,发现不适用,原因是无论是replace或on duplicate key update的原理都是:首先尝试插入,如果出现主键和唯一索引冲突,则进行更新。但问题是,哥的这个变态需求里,查询的条件既不是主键也不是唯一索引,于是这个方法1不行。

那么只能使用方法2了,使用spring的回调机制和ibatis,编码如下(下面的编码参考了http://www.cnblogs.com/sunwei2012/archive/2010/11/26/1888497.html):

public int insertOrUpdateXXDOBatch (final List< XXDO> xxDOs) throws DAOException {
       if (null == sqlMapClientTemplate.getSqlMapClient()) {
               throw new DAOException("未初始化好sqlMapClientTemplate,请检查配置及init方法" );
       }
       // 执行回调
       sqlMapClientTemplate.execute( new SqlMapClientCallback() {
               // 实现回调接口
               public Object doInSqlMapClient( SqlMapExecutor executor)
                               throws SQLException {
                       // 开始批处理
                       executor. startBatch();
                       for (XXDO xxDO : xxDOs) {
                               if (null != xxDO && null != xxDO.getCase()) {
                                       int updateCount = executor.update("IbatisDAO.update", xxDO);
                                       if (updateCount == 0 ) {
                                        executor.update("IbatisDAO.insert", xxDO);// 这里是用update方法来执行insert语句,以便于批处理
                                       }
                               }
                       }
                       // 执行批处理
                       executor. executeBatch();
                       return null;
               }
       });
       return 0 ;
}

需要注意,在ibatis中,可以使用update方法来执行insert语句。

然而,在运行时,又出现了问题:既更新了一条,又插入了一条数据;

这个问题检查一下代码就能发现,这跟ibatis的executeBatch有关,在ibatis中,批量执行实质上是将一系列sql语句组合成一个几个,然后一次性push到数据库去执行。这样既能提高执行效率,又能降低通信消耗;然而,在本代码中if(updateCount==0)这个判断其实是没有起到任何作用的,因为executeBatch是不管这个判断直接将update语句和insert语句给组合到一个集合中批量执行。

为了解决这个问题,我只能直接使用两条update语句,第一条语句直接将原来的数据逻辑delete掉,第二条insert进一条数据,代码如下:

public int insertOrUpdateXXDOBatch (final List< XXDO> xxDOs) throws DAOException {
       if (null == sqlMapClientTemplate.getSqlMapClient()) {
               throw new DAOException("未初始化好sqlMapClientTemplate,请检查配置及init方法" );
       }
       // 执行回调
       sqlMapClientTemplate.execute( new SqlMapClientCallback() {
               // 实现回调接口
               public Object doInSqlMapClient( SqlMapExecutor executor)
                               throws SQLException {
                       // 开始批处理
                       executor. startBatch();
                       for (XXDO xxDO : xxDOs) {
                               if (null != xxDO && null != xxDO.getCase()) {
                      executor.update("IbatisDAO.logicDelete" , xxDO);
                      executor.update("IbatisDAO.insert", xxDO);// 这里是用update方法来执行insert语句,以便于批处理
                               }
                       }
                       // 执行批处理
                       executor. executeBatch();
                       return null;
               }
       });
       return 0 ;
}

这仍然存在一个问题:先delete,再insert,在delete和insert的瞬间如果有读取,就会出现数据丢失。所以最好的办法,是在这里加上一个锁,不过两个方面的原因没有这么做:

业务上出现在delete之后立即读取的可能性几乎没有

本人能力暂时还不足,对于锁机制了解还不足,所以将在以后对这一方面了解的更加深入的时候,再进行优化。

至于服务超时的问题,解决方案很简单,将上面的批量执行代码写在一个Runable的实现类中,并使用下面的代码来实现一个异步线程,在服务中启动这个线程,之后立即返回,异步线程将批量执行的结果写入数据库中的一个字段里,查看结果时,从数据库中读取该结果即可。

     ExecutorService exec = Executors.newSingleThreadExecutor();
     exec.execute(new BatchUpdateThread(xxDos));
     exec.shutdown();

水平有限,欢迎留言指正。

转载请注明出处,含链接:http://www.cnblogs.com/zhguang/archive/2013/05/23/3094615.html