Mybatis中使用流式查詢避免數據量過大致使OOM

併發編程網 – ifeve.com 2017-08-03 665 閱讀

JAVA

1、前言

前面介紹了裸露JDBC 方式使用流式編程，下面介紹下MYbatis中兩種使用流式查詢方法

2、Mybaits中MyBatisCursorItemReader的使用

2.1 配置

• MyBatisCursorItemReader的注入

<bean id="myMyBatisCursorItemReader" class="org.mybatis.spring.batch.MyBatisCursorItemReader">
    <property name="sqlSessionFactory" ref="sqlSessionFactory" />
    <property name="queryId"
        value="com.taobao.accs.mass.petadata.dal.sqlmap.AccsDeviceInfoDAOMapper.selectByExampleForPetaData" />
</bean>

其中queryId爲mapper文件中接口名稱。

• Mapper.xml設置



image.png

其中fetchSize=」-2147483648″，Integer.MIN_VALUE=-2147483648

2.2 使用

static void testCursor1() throws UnexpectedInputException, ParseException, Exception {

        try {
            Map<String, Object> param = new HashMap<String, Object>();
            

            AccsDeviceInfoDAOExample accsDeviceInfoDAOExample = new AccsDeviceInfoDAOExample();
            accsDeviceInfoDAOExample.createCriteria().andAppKeyEqualTo("12345").andAppVersionEqualTo("5.7.2.4.5")
            .andPackageNameEqualTo("com.test.zlx");

            param.put("oredCriteria", accsDeviceInfoDAOExample.getOredCriteria());

            // 設置參數
            myMyBatisCursorItemReader.setParameterValues(param);
            
            // 建立遊標
            myMyBatisCursorItemReader.open(new ExecutionContext());

            //使用遊標迭代獲取每一個記錄
            Long count = 0L;
            AccsDeviceInfoDAO accsDeviceInfoDAO;
            while ((accsDeviceInfoDAO = myMyBatisCursorItemReader.read()) != null) {

                System.out.println(JSON.toJSONString(accsDeviceInfoDAO));
                ++count;
                System.out.println(count);

            }
        } catch (Exception e) {
            System.out.println("error:" + e.getLocalizedMessage());
        } finally {

            // do some
            myMyBatisCursorItemReader.close();
        }

}

 

2.3 原理簡單介紹

• open函數

做用從session工廠獲取一個session，而後調用session的selectCursor,它最終會調用

ConnectionImpl的prepareStatement方法：

public java.sql.PreparedStatement prepareStatement(String sql) throws SQLException {
    return prepareStatement(sql, DEFAULT_RESULT_SET_TYPE, DEFAULT_RESULT_SET_CONCURRENCY);
}
private static final int DEFAULT_RESULT_SET_TYPE = ResultSet.TYPE_FORWARD_ONLY;

private static final int DEFAULT_RESULT_SET_CONCURRENCY = ResultSet.CONCUR_READ_ONLY;

至此三個條件知足了兩個，在加上咱們本身設置的fetchSize就通知mysql要建立流式ResultSet。 
那麼fectchsize何處設置那？

image.png

圖中1建立prepareStatement，2設置fetchSize.

設置後最後會調用MysqlIO的sqlQueryDirect方法執行具體sql並把結果resultset存放到JDBC4PrepardStatement中。

• read函數

read函數做用是從結果集resultset中獲取數據，首先調用.next判斷是否有數據，有的話則讀取數據。 
這和純粹JDBC編程方式就同樣了，只是read函數對其進行了包裝。

3、Mybatis中ResultHandler的使用

3.1 配置

• Mapper.xml設置



image.png

其中fetchSize=」-2147483648″，Integer.MIN_VALUE=-2147483648

3.2 使用

static void testCursor2() {

    SqlSession session = sqlSessionFactory.openSession();
    Map<String, Object> param = new HashMap<String, Object>();
    
    AccsDeviceInfoDAOExample accsDeviceInfoDAOExample = new AccsDeviceInfoDAOExample();
    accsDeviceInfoDAOExample.createCriteria().andAppKeyEqualTo("12345").andAppVersionEqualTo("1.2.3.4")
            .andPackageNameEqualTo("com.hello.test");

    param.put("oredCriteria", accsDeviceInfoDAOExample.getOredCriteria());

    session.select("com.taobao.accs.mass.petadata.dal.sqlmap.AccsDeviceInfoDAOMapper.selectByExampleForPetaData",
            param, new ResultHandler() {

                @Override
                public void handleResult(ResultContext resultContext) {
                    AccsDeviceInfoDAO accsDeviceInfoDAO = (AccsDeviceInfoDAO) resultContext.getResultObject();

                    System.out.println(resultContext.getResultCount());
                    System.out.println(JSON.toJSONString(accsDeviceInfoDAO));

                }

            });

}

3.3 原理簡單介紹

相似第三節，只是第三節返回了操做ResultSet的遊標讓用戶本身迭代獲取數據，而如今是內部直接操做ResultSet逐條獲取數據並調用回調handler的handleResult方法進行處理。

private void handleRowValuesForSimpleResultMap(ResultSetWrapper rsw, ResultMap resultMap, ResultHandler<?> resultHandler, RowBounds rowBounds, ResultMapping parentMapping)
      throws SQLException {
    DefaultResultContext<Object> resultContext = new DefaultResultContext<Object>();
    skipRows(rsw.getResultSet(), rowBounds);
    while (shouldProcessMoreRows(resultContext, rowBounds) && rsw.getResultSet().next()) {
      ResultMap discriminatedResultMap = resolveDiscriminatedResultMap(rsw.getResultSet(), resultMap, null);
      Object rowValue = getRowValue(rsw, discriminatedResultMap);
      storeObject(resultHandler, resultContext, rowValue, parentMapping, rsw.getResultSet());
    }
  }

  private void storeObject(ResultHandler<?> resultHandler, DefaultResultContext<Object> resultContext, Object rowValue, ResultMapping parentMapping, ResultSet rs) throws SQLException {
    if (parentMapping != null) {
      linkToParents(rs, parentMapping, rowValue);
    } else {
      callResultHandler(resultHandler, resultContext, rowValue);
    }
  }
   
  //調用回調
  @SuppressWarnings("unchecked" /* because ResultHandler<?> is always ResultHandler<Object>*/)
  private void callResultHandler(ResultHandler<?> resultHandler, DefaultResultContext<Object> resultContext, Object rowValue) {
    resultContext.nextResultObject(rowValue);
    ((ResultHandler<Object>) resultHandler).handleResult(resultContext);
  }

4、總結與結果對比

流式編程使用裸露JDBC編程最簡單，靈活，可是sql語句須要分散寫到須要調用數據庫操做的地方，不便於維護，Mybatis底層仍是使用裸露JDBC編程API實現的，而且使用xml文件統一管理sql語句，雖然解析執行時候會有點開銷（好比每次調用都是反射進行的），可是同時還提供了緩存。

對於同等條件下搜索結果爲600萬條記錄的時候使用遊標與不使用時候內存佔用對比：

• 非流式

  image.png

• 流式

  粘貼圖片.png

可知非流式時候內存會隨着搜出來的記錄增加而近乎直線增加，流式時候則比較平穩，另外非流式因爲須要mysql服務器準備所有數據，因此調用後不會立刻返回，須要根據數據量大小不一樣會等待一段時候纔會返回，這時候調用方線程會阻塞，流式則由於每次返回一條記錄，因此返回速度會很快。

裸露JDBC流式使用參考： http://www.jianshu.com/p/c1e6eeb71c74

這裏在總結下：client發送select請求給Server後，Server根據條件篩選符合條件的記錄，而後就會把記錄發送到本身的發送buffer,等buffer滿了就flush緩存（這裏要注意的是若是client的接受緩存滿了，那麼Server的發送就會阻塞主，直到client的接受緩存空閒。），經過網絡發送到client的接受緩存，當不用遊標時候MySqIo就會從接受緩存裏面逐個讀取記錄到resultset。就這樣client 從本身的接受緩存讀取數據到resultset,同時Server端不斷經過網絡向client接受緩存發送數據，直到全部記錄都放到了resultset。

若是使用了遊標，則用戶調用resultset的next的頻率決定了Server發送時候的阻塞狀況，若是用戶調用next快，那麼client的接受緩存就會有空閒，那麼Server就會把數據發送過來，若是用戶調用的慢，那麼因爲接受緩存騰不出來，Server的發送就會阻塞