批量插入數據(基於Mybatis的實現-Oracle)

-----------------------------------

20170528 第二次編輯：主要是補充mysql相關內容。

-----------------------------------

 

mysql支持batch提交改進方案：聲明：mysql仍然沒有內部游標，讓數據庫支持executeBatch的方式處理。

MySql 的批量操作，要加rewriteBatchedStatements參數

引用“

MySql的JDBC連接的url中要加rewriteBatchedStatements參數，並保證5.1.13以上版本的驅動，才能實現高性能的批量插入。

 

例如： String connectionUrl="jdbc:mysql://192.168.1.100:3306/test?rewriteBatchedStatements=true" ; 

 

還要保證Mysql JDBC驅的版本。MySql的JDBC驅動的批量插入操作性能是很優秀的。

”

原文鏈接：http://elf8848.iteye.com/blog/770032  (建議去看看這個鏈接的評論區)

參考資料：https://www.oschina.net/question/2553117_2162171?sort=time 

20170528親測，插入26663條數據，

加上rewriteBatchedStatements后，耗時：3734毫秒
不加rewriteBatchedStatements前，耗時：672551毫秒

mysql裝在本機上，行字段數多，僅從本次測試看，性能提高了180倍。

 ---------------------------------------------------

第一版：20170526 原創

 ---------------------------------------------------

前言：做一個數據同步項目，要求：同步數據不丟失的情況下，提高插入性能。

項目DB框架：Mybatis。DataBase:Oracle。

----------------------------------------------------------------------------

批量插入數據方式：

一、Mybatis 全局設置批處理；

二、Mybatis 局部設置批處理；

三、Mybatis foreach批量插入：

①SELECT UNION ALL；

②BEGIN INSERT INTO ...;INSERT INTO...;...;END;

四、java自帶的批處理插入；

五、其他方式

-----------------------------------------------------------------------------

先說結論：Mybatis(全局/局部)批處理和java自帶的批處理 性能上差不多，屬於最優處理辦法，我這邊各種測試后，最后采用Mybatis局部批處理方式。

一、Mybatis 全局設置批處理

先上Spring-Mybatis.xml 配置信息

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context" xmlns:tx="http://www.springframework.org/schema/tx"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
                           http://www.springframework.org/schema/beans/spring-beans.xsd
                           http://www.springframework.org/schema/context
                           http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx.xsd">

    <!-- 自動掃描(自動注入) -->
    <context:annotation-config/>
    <context:component-scan base-package="com.company.dao"/>

    <!-- 動態數據源 -->
    <bean id="dataSource" class="com.company.dao.datasource.DataSource">
        <property name="myConfigFile" value="mySource.xml"/>
    </bean>

    <!-- mybatis配置 -->
    <bean id="sqlSessionFactory" class="org.mybatis.spring.SqlSessionFactoryBean">
        <property name="dataSource" ref="dataSource"/>
        <property name="mapperLocations" value="classpath*:mapper/*/*/*.xml"/>
        <property name="configLocation" value="classpath:/mybatisConfig.xml"/>
    </bean>

    <!-- 自動創建映射器，不用單獨為每個 mapper映射-->
    <bean class="org.mybatis.spring.mapper.MapperScannerConfigurer">
        <property name="basePackage" value="com.company.dao.mapper"/>
        <property name="sqlSessionFactoryBeanName" value="sqlSessionFactory"/>
    </bean>

    <!-- 事務管理器配置,單數據源事務 -->
    <bean id="transactionManager"
          class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>

    <tx:annotation-driven transaction-manager="transactionManager"/>

</beans>

再上mybatisConfig.xml（在本項目中，我沒有設置setting。最終采用的局部批處理，因此未設置全局批處理，具體原因后面再說。）

<?mapper.xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN" "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
    
    <settings>
        <!-- 配置默認的執行器。SIMPLE 就是普通的執行器；REUSE 執行器會重用預處理語句（prepared statements）； BATCH 執行器將重用語句並執行批量更新。-->
        <setting name="defaultExecutorType" value="BATCH"/>
        <!--詳見：http://www.mybatis.org/mybatis-3/zh/configuration.html-->
    </settings>
    
    <!-- 別名列表 -->
    <typeAliases>
       <!-- typeAliases 中的配置都是配置別名，在此就不貼出來了 -->
    </typeAliases>

</configuration>

這樣子設置好后，在BaseService開放saveBatch(List<T> list)方法

@Override
    public void save(List<T> list) {
        for (int i = 0;i < list.size();i++){
            mapper.insert(list.get(i));
        }
    }

    @Override
    public void saveBatch(List<T> list) {
        int size = list.size();
        int unitNum = 500;
        int startIndex = 0;
        int endIndex = 0;
        while (size > 0){
            if(size > unitNum){
                endIndex = startIndex+unitNum;
            }else {
                endIndex = startIndex+size;
            }
            List<T> insertData = list.subList(startIndex,endIndex);
            save(insertData);
            size = size - unitNum;
            startIndex = endIndex;
        }
    }

 

雖然看上去是500條記錄，一次次INSERT INTO，但由於在全局已經設置Mybatis是批處理執行器，所以這500條INSERT INTO只會與Oracle數據庫通信一次。

 

全局設置批處理的局限性在哪里呢？

先附上mybatis官方的討論列表中最很關鍵的一句：“If the BATCH executor is in use, the update counts are being lost. ”

設置全局批處理后，DB里的insert、Update和delete方法，都無法返回進行DML影響DB_TABLE的行數。

1.insert 無法返回影響的行數，這個好解決，一個批處理放在一個事務里，記錄批處理失敗次數，總數-批處理失敗次數*單位批處理數據量，就能得到insert 影響DB_TABLE的行數；

2.但是update和delete就無法很簡單的去統計影響行數了，如果做反復查詢，反而降低了效率，得不償失。

 

雖現在的項目尚未有需要反饋影響DB_TABLE行數的需求，但是為了更靈活，我們放棄了全局批處理的方式。

!這里提個疑問：為什么Mybatis官方，不將批處理的選擇方式下沉到方法級別？方便開發者根據實際情況，靈活選擇。我覺得這是個可以改進的地方，如有機會，可看源碼去進行改進。

---------------------------------------------------------------------------------------------------------

二、Mybatis局部批處理方式

 

由於領導說全局批處理方式，不夠靈活，不適宜項目所需，要另想辦法支持。但是java自帶的批處理，因為項目代碼管理的要求，也不能采用。因此，在仔細閱讀官方文檔后，設想自己能否獲取SQLSession后openSession，將這個會話設置為批處理呢？

先看MyBatis官方網站(須FanQiang)：http://www.mybatis.org/mybatis-3/zh/getting-started.html

 

SqlSession session = sqlSessionFactory.openSession();
try {
  BlogMapper mapper = session.getMapper(BlogMapper.class);
  // do work
} finally {
  session.close();
}

 

后查閱Mybatis java API(須FanQiang):  http://www.mybatis.org/mybatis-3/zh/java-api.html

現在你有一個 SqlSessionFactory,可以用來創建 SqlSession 實例。

SqlSessionFactory

SqlSessionFactory 有六個方法可以用來創建 SqlSession 實例。通常來說,如何決定是你 選擇下面這些方法時:

• Transaction (事務): 你想為 session 使用事務或者使用自動提交(通常意味着很多 數據庫和/或 JDBC 驅動沒有事務)?
• Connection (連接): 你想 MyBatis 獲得來自配置的數據源的連接還是提供你自己
• Execution (執行): 你想 MyBatis 復用預處理語句和/或批量更新語句(包括插入和 刪除)

 

重載的 openSession()方法簽名設置允許你選擇這些可選中的任何一個組合。

SqlSession openSession()
SqlSession openSession(boolean autoCommit)
SqlSession openSession(Connection connection)
SqlSession openSession(TransactionIsolationLevel level)
SqlSession openSession(ExecutorType execType,TransactionIsolationLevel level)
SqlSession openSession(ExecutorType execType)
SqlSession openSession(ExecutorType execType, boolean autoCommit)
SqlSession openSession(ExecutorType execType, Connection connection)
Configuration getConfiguration();

因此出來了局部批處理第一套代碼實現方式：

public static void sqlSession(List<Student> data) throws IOException {
        String resource = "mybatis-dataSource.xml";
        InputStream inputStream = null;
        SqlSession batchSqlSession = null;
        try{
            inputStream = Resources.getResourceAsStream(resource);
            SqlSessionFactory sqlSessionFactory = new SqlSessionFactoryBuilder().build(inputStream);
            batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
            int batchCount = 500;//每批commit的個數
            for(int index = 0; index < data.size();index++){
                Student stu = data.get(index);
                batchSqlSession.getMapper(Student.class).insert(stu);
                if(index !=0 && index%batchCount == 0){
                    batchSqlSession.commit();
                }
            }
            batchSqlSession.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(batchSqlSession != null){
                batchSqlSession.close();
            }
            if(inputStream != null){
                inputStream.close();
            }
        }
    }

<?xml version="1.0" encoding="UTF-8" ?>
<!DOCTYPE configuration
  PUBLIC "-//mybatis.org//DTD Config 3.0//EN"
  "http://mybatis.org/dtd/mybatis-3-config.dtd">
<configuration>
  <environments default="development">
    <environment id="development">
      <transactionManager type="JDBC"/>
      <dataSource type="POOLED">
        <property name="driver" value="${driver}"/>
        <property name="url" value="${url}"/>
        <property name="username" value="${username}"/>
        <property name="password" value="${password}"/>
      </dataSource>
    </environment>
  </environments>
  <mappers>
    <mapper resource="org/mybatis/example/Student.xml"/>
  </mappers>
</configuration>

已經在Spring-Mybatis.xml 中配置了SQLSessionFactory，那我為何還要自己去創建SQLSessionFactory呢？因此繼續改良代碼

public static void mybatisSqlSession(List<Student> data){
        DefaultSqlSessionFactory sqlSessionFactory = (DefaultSqlSessionFactory) ServiceBeanConstant.CTX.getBean("sqlSessionFactory");
        SqlSession batchSqlSession = null;
        try{
            batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
            int batchCount = 500;//每批commit的個數
            for(int index = 0; index < data.size();index++){
                Student stu = data.get(index);
                batchSqlSession.getMapper(StudentMapper.class).insert(stu);
                if(index !=0 && index%batchCount == 0){
                    batchSqlSession.commit();
                }
            }
            batchSqlSession.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(batchSqlSession != null){
                batchSqlSession.close();
            }
        }
    }

 

這個版本的局部批處理插入是比較滿意的，最終采用的方式也是這個版本。

 

下面放出在IService接口定義和Service的具體實現代碼：

IService接口定義

/**
     * 批處理插入數據(方法內部定義500條為一個批次進行提交)
     * 使用注意事項：必須在XxxMappper.xml中實現<insert id="insert" ...>....<insert/>的sql
     * @param data 批量插入的數據
     * @param mClass 調用的XxxMaperr.class
     * @auth robin
     * Created on 2016/3/14
     */
    void saveBatch(List<T> data,Class mClass);

Service實現

@Override
    public void saveBatch(List<T> data,Class mClass) {
        DefaultSqlSessionFactory sqlSessionFactory = (DefaultSqlSessionFactory) ServiceBeanConstant.CTX.getBean("sqlSessionFactory");
        SqlSession batchSqlSession = null;
        try{
            batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
            int batchCount = 500;//每批commit的個數
            for(int index = 0; index < data.size();index++){
                T t = data.get(index);
                ((BaseMapper)batchSqlSession.getMapper(mClass)).insert(t);
                if(index !=0 && index%batchCount == 0){
                    batchSqlSession.commit();
                }
            }
            batchSqlSession.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(batchSqlSession != null){
                batchSqlSession.close();
            }
        }
    }

 

局部和全局批處理插入對比：局部批處理，可以對特定一類的方法，進行數據批處理，不會影響其他DML語句，其他DML語句，可以正常返回影響DB_TABLE的行數。

!這樣既能針對特殊需求(批處理)支持，也能支持未來需要返回影響數據行的要求。

 

注意：使用批處理方式進行DML操作，是無法反饋影響DB_TABLE行數的數據。無論是局部批處理還是java自帶的批處理方式，皆無法反饋DB_TABLE count。

 

補充完善：

在我的Service實現中，通過注入的方式，獲取mapper的實例

 

public class BaseService<MAPPER extends BaseMapper, T, PK extends Serializable> implements IBaseService<T, PK> {

    protected T tt;
    /**
     * 實體操作的自動注入Mapper(隨初始化一同注入，必須用set方法)
     */
    protected MAPPER mapper;

    public MAPPER getMapper() {
        return mapper;
    }

    @Autowired
    public void setMapper(MAPPER mapper) {
        this.mapper = mapper;
    }
   //后續代碼略
}

 

前面的Service saveBatch方法中，還需要傳入指定的Mapper.class.對本項目其他開發者來說，與之前的環境相比，多傳一個參數感覺別扭。

那么為何我不繼續封裝，外部無需傳入Mapper.class，而是通過內部注入的mapper實例獲取Mapper.class.
改良后的代碼：

 

@Override
    public T saveBatch(List<T> data) {
        T back = null;
        DefaultSqlSessionFactory sqlSessionFactory = (DefaultSqlSessionFactory) ServiceBeanConstant.CTX.getBean("sqlSessionFactory");
        SqlSession batchSqlSession = null;
        try{
            batchSqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH, false);
            int batchCount = 500;//每批commit的個數
            for(int index = 0; index < data.size();index++){
                T t = data.get(index);
                back = t;
                Class<?>[] interfaces=mapper.getClass().getInterfaces();
                Class clazz = null;
                for (int i=0;i<interfaces.length;i++){
                    if(BaseMapper.class.isAssignableFrom(interfaces[i])){
                        clazz = interfaces[i];
                    }
                }
                if(clazz == null){
                    throw new Exception("user-defined exception:mapper not implements interfaces com.company.dao.mapper.BaseMapper");
                }
                BaseMapper baseMapper = (BaseMapper) batchSqlSession.getMapper(clazz);
                baseMapper.insert(t);
                if(index !=0 && index%batchCount == 0){
                    batchSqlSession.commit();
                }
            }
            batchSqlSession.commit();
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            if(batchSqlSession != null){
                batchSqlSession.close();
            }
            return back;
        }
    }

 

這里對mapper實例進行一個簡短的說明：

1.mapper實例是通過java動態代理來實例化的；

2.mapper的SQLSession是使用mybatis統一的配置實例的；

3.mapper的默認執行器是SIMPLE（普通的執行器）;

-------------------------------------------------------------------------------------

 三、Mybatis foreach批量插入

Mybatis foreach 批量插入，如果批量插入的數據量大，不得不說這真是一個非常糟糕的做法。

無論是SELECT ** UNION ALL 還是BEGIN ...;END; ，相對而言后者比前者稍微好點。

放出DB和我測試的結果：

耗時	占當時整個數據庫CPU百分比	說明
15.5	98.33	union all方式拼接插入
16.4	97.75	begin end方式插入塊
1.54	64.81	java 自帶的batch方式插入

①foreach union all的批量插入，現已有大量的博客資源可供參考，我就不貼出自己的實現方式了。

如果有興趣可以參閱：http://blog.csdn.net/sanyuesan0000/article/details/19998727 (打開瀏覽器，復制url)

這篇博客。BEGIN END的方式，也是從這篇博客中得到啟發。只不過他是把BEGIN END用在update中。

②foreach begin end 語句塊

我的實現：

<insert id="insertBatch" parameterType="java.util.List">
           BEGIN
           <foreach collection="list" item="item" index="index" separator=";" >
               INSERT INTO TABLE.STUDENT (ID,AGE,NAME,STU_ID) VALUES
               ( DEMO.SEQ_EID.NEXTVAL,#{item.age},#{item.name},#{item.stuId} )
           </foreach>
           ;END ;
       </insert>

 調用方式：

@Override
    public void saveBatch(List<T> list) {
        int size = list.size();
        int unitNum = 500;
        int startIndex = 0;
        int endIndex = 0;
        while (size > 0){
            if(size > unitNum){
                endIndex = startIndex+unitNum;
            }else {
                endIndex = startIndex+size;
            }
            List<T> insertData = list.subList(startIndex,endIndex);
            mapper.insertBatch(insertData);
            size = size - unitNum;
            startIndex = endIndex;
        }

---------------------------------------------------------------------

四、java自帶的批處理方式

廢話不多說，直接上代碼

package DB;

import base.Student;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by robin on 2016/5/23.
 *
 * @author robin
 */
public class InsertTableDemo {

    public static void main(String args[]) throws SQLException {
        Connection connection = null;
        List<Student> dataList = getDataList(100000);
        long startTime = 0;
        try{
            connection = getConn();
            startTime=System.currentTimeMillis();
            connection.setAutoCommit(false);
            PreparedStatement statement = connection.prepareStatement("INSERT INTO STUDENT (ID,AGE,NAME,STU_ID) VALUES ( DEMO.SEQ_EID.NEXTVAL, ?,?,? ) ");
            int num = 0;
            for (int i = 0;i< dataList.size();i++){
                Student s = dataList.get(i);
                statement.setInt(1, s.getAge());
                statement.setString(2, s.getName());
                statement.setString(3, s.getStuId());
                statement.addBatch();
                num++;
                if(num !=0 && num%500 == 0){
                    statement.executeBatch();
                    connection.commit();
                    num = 0;
                }
            }
            statement.executeBatch();
            connection.commit();
        }catch (Exception e){
            e.printStackTrace();
            connection.rollback();
        }finally {
            if(connection != null){
                connection.close();
            }
            long endTime=System.currentTimeMillis();
            System.out.println("方法執行時間："+(endTime-startTime)+"ms");
        }

    }

    public static Connection getConn(){
        String driver = "oracle.jdbc.driver.OracleDriver";
        String url = "jdbc:oracle:thin:@//ip:port/DMEO"; //DMEO為數據庫名
        String user = "user";
        String password = "pwd";
        try{
            Class.forName(driver);
            Connection conn = DriverManager.getConnection(url, user, password);
            return conn;
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }
    public static List<Student> getDataList(int f){
        List<Student> data = new ArrayList<>();
        for (int i =0;i<f;i++){
            Student s = new Student(""+i,"小明" + i,i);
            data.add(s);
        }
        return data;
    }


}

這種批量插入大量數據的方式，性能上最好。但是因為我們小組代碼管理所限制，因此這種方式不使用。

------------------------------------------------------------------------

五、其他方式

現在已經忘了，其他方式到底使用過哪些，但總歸是比以上四種效果都更差，所以沒什么印象了。

如果各位，還有什么其他更好的批量插入數據的方式，歡迎加入討論，集思廣益。 

 

以上就是這兩天，對在原項目基礎上不進行大變動的基礎上，提供批處理插入數據的所思所行。

-------------------------------------------------------------------------------

后記：

 

這里吐槽一句：希望大家不要把未經過自己驗證的東西，言之鑿鑿地寫到博客中去。

在我做批處理這件事的時候，領導也在參閱網上的博客。其中有一篇博客，說自己在oracle中批量插入數據，采用foreach insert into (field1,field2,...) values (v11,v12,...),(v21,v22,...) ,(v31,v32,...),...也可以。

雖然我明知不行，但是無可奈何還是要去演示給領導看，在oracle中，這種寫法確實不適用。

領導問我為何他說可以，我想我也只能回答：他抄別人的博客唄，抄來抄去都不自己實踐驗證，就想當然地寫到博客里。

 

所以，如果你看完了我這篇分享，希望您也能寫個demo驗證下，起碼可以加深自己的理解。

 

 感謝領導和DB同事，在此過程中的幫助。

以上內容，都經過本人實踐驗證過。若轉發，請在標題上標記[轉]，並注明原文鏈接：http://www.cnblogs.com/robinjava77/p/5530681.html，作者名稱：robin。並在文章最后一行附上本句話。否則，作者保留追究的權利。