炸！使用 MyBatis 查詢千萬數(shù)據(jù)量？

作者：馬稱 2021-01-07 07:46:34

由于現(xiàn)在 ORM 框架的成熟運(yùn)用，很多小伙伴對于 JDBC 的概念有些薄弱，ORM 框架底層其實(shí)是通過 JDBC 操作的 DB。

本文轉(zhuǎn)載自微信公眾號「源碼興趣圈」，作者馬稱。轉(zhuǎn)載本文請聯(lián)系源碼興趣圈公眾號。

由于現(xiàn)在 ORM 框架的成熟運(yùn)用，很多小伙伴對于 JDBC 的概念有些薄弱，ORM 框架底層其實(shí)是通過 JDBC 操作的 DB

JDBC(JavaDataBase Connectivity)是 Java 數(shù)據(jù)庫連接, 說的直白點(diǎn)就是使用 Java 語言操作數(shù)據(jù)庫

由 SUN 公司提供出一套訪問數(shù)據(jù)庫的規(guī)范 API，并提供相對應(yīng)的連接數(shù)據(jù)庫協(xié)議標(biāo)準(zhǔn)，然后各廠商根據(jù)規(guī)范提供一套訪問自家數(shù)據(jù)庫的 API 接口

文章大數(shù)據(jù)量操作核心圍繞 JDBC 展開，目錄結(jié)構(gòu)如下：

MySQL JDBC 大數(shù)據(jù)量操作
- 常規(guī)查詢
- 流式查詢
- 游標(biāo)查詢
- JDBC RowData
- JDBC 通信原理
流式游標(biāo)內(nèi)存分析
- 單次調(diào)用內(nèi)存使用
- 并發(fā)調(diào)用內(nèi)存使用
MyBatis 如何使用流式查詢
結(jié)言

MySql JDBC 大數(shù)據(jù)量操作

整篇文章以大數(shù)據(jù)量操作為議題，通過開發(fā)過程中的需求引出相關(guān)知識點(diǎn)

遷移數(shù)據(jù)
導(dǎo)出數(shù)據(jù)
批量處理數(shù)據(jù)

一般而言筆者認(rèn)為在 Java Web 程序里，能夠被稱為大數(shù)據(jù)量的，幾十萬到千萬不等，再高的話 Java(WEB 應(yīng)用)處理就不怎么合適了

舉個例子，現(xiàn)在業(yè)務(wù)系統(tǒng)需要從 MySQL 數(shù)據(jù)庫里讀取 500w 數(shù)據(jù)行進(jìn)行處理，應(yīng)該怎么做

常規(guī)查詢，一次性讀取 500w 數(shù)據(jù)到 JVM 內(nèi)存中，或者分頁讀取
流式查詢，建立長連接，利用服務(wù)端游標(biāo)，每次讀取一條加載到 JVM 內(nèi)存
游標(biāo)查詢，和流式一樣，通過 fetchSize 參數(shù)，控制一次讀取多少條數(shù)據(jù)

常規(guī)查詢

默認(rèn)情況下，完整的檢索結(jié)果集會將其存儲在內(nèi)存中。在大多數(shù)情況下，這是最有效的操作方式，并且由于 MySQL 網(wǎng)絡(luò)協(xié)議的設(shè)計，因此更易于實(shí)現(xiàn)

假設(shè)單表 500w 數(shù)據(jù)量，沒有人會一次性加載到內(nèi)存中，一般會采用分頁的方式

@SneakyThrows 
@Override 
public void pageQuery() { 
    @Cleanup Connection conn = dataSource.getConnection(); 
    @Cleanup Statement stmt = conn.createStatement(); 
    long start = System.currentTimeMillis(); 
    long offset = 0; 
    int size = 100; 
    while (true) { 
        String sql = String.format("SELECT COLUMN_A, COLUMN_B, COLUMN_C FROM YOU_TABLE LIMIT %s, %s", offset, size); 
        @Cleanup ResultSet rs = stmt.executeQuery(sql); 
        long count = loopResultSet(rs); 
        if (count == 0) break; 
        offset += size; 
    } 
 
    log.info("  🚀🚀🚀 分頁查詢耗時 :: {} ", System.currentTimeMillis() - start); 
}

上述方式比較簡單，但是在不考慮 LIMIT 深分頁優(yōu)化情況下，線上數(shù)據(jù)庫服務(wù)器就涼了，亦或者你能等個幾天時間檢索數(shù)據(jù)

流式查詢

如果你正在使用具有大量數(shù)據(jù)行的 ResultSet，并且無法在 JVM 中為其分配所需的內(nèi)存堆空間，則可以告訴驅(qū)動程序從結(jié)果流中返回一行

流式查詢有一點(diǎn)需要注意：必須先讀取(或關(guān)閉)結(jié)果集中的所有行，然后才能對連接發(fā)出任何其他查詢，否則將引發(fā)異常

使用流式查詢，則要保持對產(chǎn)生結(jié)果集的語句所引用的表的并發(fā)訪問，因?yàn)槠?查詢會獨(dú)占連接，所以必須盡快處理

@SneakyThrows 
public void streamQuery() { 
    @Cleanup Connection conn = dataSource.getConnection(); 
    @Cleanup Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); 
    stmt.setFetchSize(Integer.MIN_VALUE); 
 
   long start = System.currentTimeMillis(); 
    @Cleanup ResultSet rs = stmt.executeQuery("SELECT COLUMN_A, COLUMN_B, COLUMN_C FROM YOU_TABLE"); 
    loopResultSet(rs); 
    log.info("  🚀🚀🚀 流式查詢耗時 :: {} ", (System.currentTimeMillis() - start) / 1000); 
}

流式查詢庫表數(shù)據(jù)量 500w 單次調(diào)用時間消耗：≈ 6s

游標(biāo)查詢

SpringBoot 2.x 版本默認(rèn)連接池為 HikariPool，連接對象是 HikariProxyConnection，所以下述設(shè)置游標(biāo)方式就不可行了

((JDBC4Connection) conn).setUseCursorFetch(true);

需要在數(shù)據(jù)庫連接信息里拼接 &useCursorFetch=true。其次設(shè)置 Statement 每次讀取數(shù)據(jù)數(shù)量，比如一次讀取 1000

@SneakyThrows 
public void cursorQuery() { 
    @Cleanup Connection conn = dataSource.getConnection(); 
    @Cleanup Statement stmt = conn.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); 
    stmt.setFetchSize(1000); 
 
    long start = System.currentTimeMillis(); 
    @Cleanup ResultSet rs = stmt.executeQuery("SELECT COLUMN_A, COLUMN_B, COLUMN_C FROM YOU_TABLE"); 
    loopResultSet(rs); 
    log.info("  🚀🚀🚀 游標(biāo)查詢耗時 :: {} ", (System.currentTimeMillis() - start) / 1000); 
}

游標(biāo)查詢庫表數(shù)據(jù)量 500w 單次調(diào)用時間消耗：≈ 18s

JDBC RowData

上面都使用到了方法 loopResultSet，方法內(nèi)部只是進(jìn)行了 while 循環(huán)，常規(guī)、流式、游標(biāo)查詢的核心點(diǎn)在于 next 方法

@SneakyThrows 
private Long loopResultSet(ResultSet rs) { 
    while (rs.next()) { 
    // 業(yè)務(wù)操作 
    } 
    return xx; 
}

ResultSet.next() 的邏輯是實(shí)現(xiàn)類 ResultSetImpl 每次都從 RowData 獲取下一行的數(shù)據(jù)。RowData 是一個接口，實(shí)現(xiàn)關(guān)系圖如下

默認(rèn)情況下 ResultSet 會使用 RowDataStatic 實(shí)例，在生成 RowDataStatic 對象時就會把 ResultSet 中所有記錄讀到內(nèi)存里，之后通過 next() 再一條條從內(nèi)存中讀

RowDataCursor 的調(diào)用為批處理，然后進(jìn)行內(nèi)部緩存，流程如下：

首先會查看自己內(nèi)部緩沖區(qū)是否有數(shù)據(jù)沒有返回，如果有則返回下一行
如果都讀取完畢，向 MySQL Server 觸發(fā)一個新的請求讀取 fetchSize 數(shù)量結(jié)果
并將返回結(jié)果緩沖到內(nèi)部緩沖區(qū)，然后返回第一行數(shù)據(jù)

當(dāng)采用流式處理時，ResultSet 使用的是 RowDataDynamic 對象，而這個對象 next() 每次調(diào)用都會發(fā)起 IO 讀取單行數(shù)據(jù)

總結(jié)來說就是，默認(rèn)的 RowDataStatic 讀取全部數(shù)據(jù)到客戶端內(nèi)存中，也就是我們的 JVM;RowDataCursor 一次讀取 fetchSize 行，消費(fèi)完成再發(fā)起請求調(diào)用;RowDataDynamic 每次 IO 調(diào)用讀取一條數(shù)據(jù)

JDBC 通信原理

普通查詢

在 JDBC 與 MySQL 服務(wù)端的交互是通過 Socket 完成的，對應(yīng)到網(wǎng)絡(luò)編程，可以把 MySQL 當(dāng)作一個 SocketServer，因此一個完整的請求鏈路應(yīng)該是：

JDBC 客戶端 -> 客戶端 Socket -> MySQL -> 檢索數(shù)據(jù)返回 -> MySQL 內(nèi)核 Socket 緩沖區(qū) -> 網(wǎng)絡(luò) -> 客戶端 Socket Buffer -> JDBC 客戶端

普通查詢的方式在查詢大數(shù)據(jù)量時，所在 JVM 可能會涼涼，原因如下：

MySQL Server 會將檢索出的 SQL 結(jié)果集通過輸出流寫入到內(nèi)核對應(yīng)的 Socket Buffer
內(nèi)核緩沖區(qū)通過 JDBC 發(fā)起的 TCP 鏈路進(jìn)行回傳數(shù)據(jù)，此時數(shù)據(jù)會先進(jìn)入 JDBC 客戶端所在內(nèi)核緩沖區(qū)
JDBC 發(fā)起 SQL 操作后，程序會被阻塞在輸入流的 read 操作上，當(dāng)緩沖區(qū)有數(shù)據(jù)時，程序會被喚醒進(jìn)而將緩沖區(qū)數(shù)據(jù)讀取到 JVM 內(nèi)存中
MySQL Server 會不斷發(fā)送數(shù)據(jù)，JDBC 不斷讀取緩沖區(qū)數(shù)據(jù)到 Java 內(nèi)存中，雖然此時數(shù)據(jù)已到 JDBC 所在程序本地，但是 JDBC 還沒有對 execute 方法調(diào)用處進(jìn)行響應(yīng)，因?yàn)樾枰鹊綄?yīng)數(shù)據(jù)讀取完畢才會返回
弊端就顯而易見了，如果查詢數(shù)據(jù)量過大，會不斷經(jīng)歷 GC，然后就是內(nèi)存溢出

游標(biāo)查詢

通過上文得知，游標(biāo)可以解決普通查詢大數(shù)據(jù)量的內(nèi)存溢出問題，但是

小伙伴有沒有思考過這么一個問題，MySQL 不知道客戶端程序何時消費(fèi)完成，此時另一連接對該表造成 DML 寫入操作應(yīng)該如何處理?

其實(shí)，在我們使用游標(biāo)查詢時，MySQL 需要建立一個臨時空間來存放需要被讀取的數(shù)據(jù)，所以不會和 DML 寫入操作產(chǎn)生沖突

但是游標(biāo)查詢會引發(fā)以下現(xiàn)象：

IOPS 飆升，因?yàn)樾枰祷氐臄?shù)據(jù)需要寫入到臨時空間中，存在大量的 IO 讀取和寫入，此流程可能會引起其它業(yè)務(wù)的寫入抖動
磁盤空間飆升，因?yàn)閷懭肱R時空間的數(shù)據(jù)是在原表之外的，如果表數(shù)據(jù)過大，極端情況下可能會導(dǎo)致數(shù)據(jù)庫磁盤寫滿，這時網(wǎng)絡(luò)輸出時沒有變化的。而寫入臨時空間的數(shù)據(jù)會在讀取完成或客戶端發(fā)起 ResultSet#close 操作時由 MySQL 回收
客戶端 JDBC 發(fā)起 SQL 查詢，可能會有長時間等待 SQL 響應(yīng)，這段時間為服務(wù)端準(zhǔn)備數(shù)據(jù)階段。但是普通查詢等待時間與游標(biāo)查詢等待時間原理上是不一致的，前者是一致在讀取網(wǎng)絡(luò)緩沖區(qū)的數(shù)據(jù)，沒有響應(yīng)到業(yè)務(wù)層面;后者是 MySQL 在準(zhǔn)備臨時數(shù)據(jù)空間，沒有響應(yīng)到 JDBC
數(shù)據(jù)準(zhǔn)備完成后，進(jìn)行到傳輸數(shù)據(jù)階段，網(wǎng)絡(luò)響應(yīng)開始飆升，IOPS 由"讀寫"轉(zhuǎn)變?yōu)?quot;讀取"

采用游標(biāo)查詢的方式通信效率比較低，因?yàn)榭蛻舳讼M(fèi)完 fetchSize 行數(shù)據(jù)，就需要發(fā)起請求到服務(wù)端請求，在數(shù)據(jù)庫前期準(zhǔn)備階段 IOPS 會非常高，占用大量的磁盤空間以及性能

流式查詢

當(dāng)客戶端與 MySQL Server 端建立起連接并且交互查詢時，MySQL Server 會通過輸出流將 SQL 結(jié)果集返回輸出，也就是向本地的內(nèi)核對應(yīng)的 Socket Buffer 中寫入數(shù)據(jù)，然后將內(nèi)核中的數(shù)據(jù)通過 TCP 鏈路回傳數(shù)據(jù)到 JDBC 對應(yīng)的服務(wù)器內(nèi)核緩沖區(qū)

JDBC 通過輸入流 read 方法去讀取內(nèi)核緩沖區(qū)數(shù)據(jù)，因?yàn)殚_啟了流式讀取，每次業(yè)務(wù)程序接收到的數(shù)據(jù)只有一條
MySQL 服務(wù)端會向 JDBC 代表的客戶端內(nèi)核源源不斷的輸送數(shù)據(jù)，直到客戶端請求 Socket 緩沖區(qū)滿，這時的 MySQL 服務(wù)端會阻塞
對于 JDBC 客戶端而言，數(shù)據(jù)每次讀取都是從本機(jī)器的內(nèi)核緩沖區(qū)，所以性能會更快一些，一般情況不必?fù)?dān)心本機(jī)內(nèi)核無數(shù)據(jù)消費(fèi)(除非 MySQL 服務(wù)端傳遞來的數(shù)據(jù)，在客戶端不做任何業(yè)務(wù)邏輯，拿到數(shù)據(jù)直接放棄，會發(fā)生客戶端消費(fèi)比服務(wù)端超前的情況)

看起來，流式要比游標(biāo)的方式更好一些，但是事情往往不像表面上那么簡單

相對于游標(biāo)查詢，流式對數(shù)據(jù)庫的影響時間要更長一些
另外流式查詢依賴網(wǎng)絡(luò)，導(dǎo)致網(wǎng)絡(luò)擁塞可能性較大

流式游標(biāo)內(nèi)存分析

表數(shù)據(jù)量：500w

內(nèi)存查看工具：JDK 自帶 Jvisualvm

設(shè)置 JVM 參數(shù)：-Xmx512m -Xms512m

單次調(diào)用內(nèi)存使用

流式查詢內(nèi)存性能報告如下

圖1 數(shù)據(jù)僅供參考

游標(biāo)查詢內(nèi)存性能報告如下

圖2 數(shù)據(jù)僅供參考

根據(jù)內(nèi)存占用情況來看，游標(biāo)查詢和流式查詢都能夠很好的防止 OOM

并發(fā)調(diào)用內(nèi)存使用

并發(fā)調(diào)用：Jmete 1 秒 10 個線程并發(fā)調(diào)用

流式查詢內(nèi)存性能報告如下

圖3 數(shù)據(jù)僅供參考

并發(fā)調(diào)用對于內(nèi)存占用情況也很 OK，不存在疊加式增加

流式查詢并發(fā)調(diào)用時間平均消耗：≈ 55s

游標(biāo)查詢內(nèi)存性能報告如下

圖4 數(shù)據(jù)僅供參考

游標(biāo)查詢并發(fā)調(diào)用時間平均消耗：≈ 83s

因?yàn)樵O(shè)備限制，以及部分情況只會在極端下產(chǎn)生，所以沒有進(jìn)行生產(chǎn)、測試多環(huán)境驗(yàn)證，小伙伴感興趣可以自行測試

MyBatis 如何使用流式查詢

上文都是在描述如何使用 JDBC 原生 API 進(jìn)行查詢，ORM 框架 Mybatis 也針對流式查詢進(jìn)行了封裝

ResultHandler 接口只包含 handleResult 方法，可以獲取到已轉(zhuǎn)換后的 Java 實(shí)體類

@Slf4j 
@Service 
public class MyBatisStreamService { 
    @Resource 
    private MyBatisStreamMapper myBatisStreamMapper; 
 
    public void mybatisStreamQuery() { 
        long start = System.currentTimeMillis(); 
        myBatisStreamMapper.mybatisStreamQuery(new ResultHandler<YOU_TABLE_DO>() { 
            @Override 
            public void handleResult(ResultContext<? extends YOU_TABLE_DO> resultContext) { } 
        }); 
        log.info("  🚀🚀🚀 MyBatis查詢耗時 :: {} ", System.currentTimeMillis() - start); 
    } 
}

除了下述注解式的應(yīng)用方式，也可以使用 .xml 文件的形式

@Mapper 
public interface MyBatisStreamMapper { 
    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = Integer.MIN_VALUE) 
    @ResultType(YOU_TABLE_DO.class) 
    @Select("SELECT COLUMN_A, COLUMN_B, COLUMN_C FROM YOU_TABLE") 
    void mybatisStreamQuery(ResultHandler<YOU_TABLE_DO> handler); 
}