MySQL 流式查詢的奧秘與應用解析
很久沒寫開篇了，針對大數據采集分析和導出等功能，我們必須在內存和性能上做好折中，這其中筆者最愛的就是流式查詢，而本文將基于多個角度針對流式查詢這技能進行深入的分析和演示，希望對你有幫助。
一、詳解流式查詢
1. 關于IOPS和數據吞吐量
為了保證后文講解的流暢，我們這里對幾個比較重要的性能指標進行簡單的科普，對于服務器系統層面，IOPS(Input/Output Per Second)磁盤每秒的讀寫次數，一般以每秒輸入輸出量為單位進行衡量。而吞吐量更多的是反應的是每秒處理的IO請求，兩者關系我們可以通過如下兩個場景了解一下差異：
- 假設我們讀取1000個1kb的數據，耗時10s，那么這個服務器的數據吞吐量100KB/s，IOPS就是100，這種場景更追求IOPS。
- 假設我們只有1個請求去讀取10M的文件，耗時0.2s，那么這個服務器的數據吞吐量就是50MB/s，IOPS為5，這個服務器就更偏向于吞吐量。
2. MySQL常見的幾種查詢
日常針對大表數據采集導出的功能，我們一般會采用一下幾種方案：
- 一次性全量導出
- 使用分頁查詢
- 使用游標查詢
- 流式查詢
(1) 全量導出
我們先來說說全量查詢，這種方案本質原理就是一次性將結果集從MySQL服務端寫到客戶端程序上，針對大表數據檢索，如果我們的程序沒有足夠的堆內存空間，存在內存溢出的風險：
(2) 分頁查詢
為了解決OOM問題，我們會考慮通過分頁查詢的方式，通過分批處理完成批量數據檢索導出的工作，這種方式雖然很好的節約了堆內存空間，但這種方案在代碼實現層面就已經非常復雜了，開發者必須考慮：
- 分頁計算(這一步就涉及數據掃描，開銷大)
- 基于分頁評估每次分頁大小
- 基于頁數進行循環查詢
- 查詢SQL需要針對深分頁問題進行優化
這種方案相較于前者雖然節省了堆內存空間且可以一定程度上避免頻繁的Full GC，對于開發者整體素質要求較高，并且這種方案在性能表現上也不是很出色：
(3) 游標查詢
所以為了避免在開發層面進行手動分頁實現的復雜度，我們就想到通過游標法進行查詢，游標也就是cursor，這種查詢方式要求客戶端一次性指明fetchSize，然后服務端每次都基于給定的fetchSize將數據寫給客戶端，直到客戶端將所有數據都處理完成。
需要了解的是游標查詢這種方案考慮到客戶端未知的處理效率，為保證服務端能夠一次性將fetch的數據寫回到客戶端，MySQL服務端會為了這個查詢建立一個臨時空間來緩存數據，在極端情況下因為這些問題：
- IOPS飆升
- 磁盤空間飆升(因為臨時空間無法在緩存中容納，寫入到文件中)
- fetch設置過大，SQL查詢經常處于阻塞等待IO數據的情況
(4) 流式查詢
最后我們就來說說本文的重點——流式查詢，當客戶端向服務端發送SQL請求后，流式查詢會得到一個迭代器，客戶端不斷通過ResultSet.next()獲取下一條數據，服務端會按照客戶端接受速率并基于迭代器的偏移量逐步寫入到網絡buffer中讓客戶端讀取，這種方式很好的解決游標查詢逐批次緩存的問題。 但需要注意的是這種方案和上述游標查詢一樣，會因為數據量的問題，使得連接長時間被當前線程持有：
二、流式查詢使用示例
接下來筆者就以常見的ORM框架Mybatis演示一下如何使用流式查詢，假設我們需要查詢一張user表，對應的我們基于Options注解給出當前這個查詢信息告知查詢resultSetType 為只讀，并且指明fetchSize 為MIN_VALUE。同時，看到筆者在方法上給出了一個ResultHandler，這個處理用于處理流式查詢響應結果后的回調處理：
@Select("select * from user ")
@Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = Integer.MIN_VALUE)
@ResultType(User.class)
void selectListByStream(ResultHandler<User> handler);需要補充說明的是上述三個配置都必須明確按照要求進行配置，原因是在mybatis在執行SQL查詢時，StatementImpl會通過createStreamingResultSet判斷這三個參數是否符合要求，只有明確符合要求返回true，后續的結果集才能被創建為ResultsetRowsStreaming：
protected boolean createStreamingResultSet() {
return ((this.query.getResultType() == Type.FORWARD_ONLY) && (this.resultSetConcurrency == java.sql.ResultSet.CONCUR_READ_ONLY)
&& (this.query.getResultFetchSize() == Integer.MIN_VALUE));
}對應的我們也給出最后的使用示例，這里筆者用lambda精簡了一下ResultHandler的聲明，每當我們收到流式響應數據后，直接獲取user并自增一下原子類：
userMapper.selectListByStream(resultContext -> {
User user = resultContext.getResultObject();
//執行業務操作......
});三、性能壓測
簡單介紹了一下流式查詢的基礎配置和使用之后，我們不妨針對上述方案進行性能和內存使用情況壓測，首先筆者已經準備了100w條數據并將堆內存設置為512M：
-Xmx512m -Xms512m我們先給出一個基于全量查詢的導出寫入到本地txt文件：
try (BufferedWriter outputStream = FileUtil.getWriter("F://tmp/userData.txt", Charset.defaultCharset(), false)) {
//全量查詢
List<User> userList = userMapper.selectList(Wrappers.emptyWrapper());
//遍歷寫入本地文件
userList.forEach(user -> {
try {
outputStream.write(JSONUtil.toJsonStr(user) + "\r\n");
count.getAndIncrement();
} catch (IOException e) {
throw new RuntimeException(e);
}
});
} catch (Exception e) {
throw new RuntimeException(e);
}最終100w數據導出跑了大約20s:
save count:1000000 cost:20382ms我們通過jstat指令查看堆內存使用情況，觸發了6次full gc，整體回收花費了6s，性能表現非常差勁：
S0C S1C S0U S1U EC EU OC OU MC MU CCSC CCSU YGC YGCT FGC FGCT GCT
512.0 512.0 0.0 64.0 173568.0 68832.5 349696.0 194931.1 44032.0 41197.8 5888.0 5351.0 95 0.614 13 6.255 6.869然后就是分頁查詢，可以看到筆者這里并沒有針對深分頁問題進行優化，明確頁數和分批數后直接進行分批查詢導出了：
//定位總數
Long c = userMapper.selectCount(Wrappers.emptyWrapper());
//計算頁數
int size = 1_0000;
long pageSize = c % size == 0 ? c / size : c / size + 1;
//分頁查詢
try (BufferedWriter outputStream = FileUtil.getWriter("F://tmp/userData.txt", Charset.defaultCharset(), false)) {
for (int i = 1; i <= pageSize; i++) {
PageHelper.startPage(i, size, false);
userMapper.selectList(Wrappers.emptyWrapper()).stream()
.forEach(u -> {
try {
outputStream.write(JSONUtil.toJsonStr(u) + "\r\n");
} catch (IOException e) {
throw new RuntimeException(e);
}
count.incrementAndGet();
});
}
} catch (Exception e) {
throw new RuntimeException(e);
}可以看到這段代碼內存使用情況比較穩定，但是耗時大約29s:
jstat -gc 21524
S0C S1C S0U S1U EC EU OC OU MC MU CCSC CCSU YGC YGCT FGC FGCT GCT
3584.0 3584.0 0.0 1831.1 167424.0 64921.0 349696.0 22386.0 44288.0 41324.7 5888.0 5362.6 81 0.283 2 0.072 0.355最后我們給出流式查詢的代碼拉取數據并寫入本地文件的示例：
try (BufferedWriter outputStream = FileUtil.getWriter("F://tmp/userData.txt", Charset.defaultCharset(), false)) {
//流式查詢
userMapper.selectListByStream(res -> {
//定位對象
User user = res.getResultObject();
try {
//寫入文件
outputStream.write(JSONUtil.toJsonStr(user) + "\r\n");
} catch (IOException e) {
throw new RuntimeException(e);
}
count.incrementAndGet();
});
} catch (Exception e) {
throw new RuntimeException(e);
}最終整體耗時14s，使用jstat查看gc情況也是非常穩定：
S0C S1C S0U S1U EC EU OC OU MC MU CCSC CCSU YGC YGCT FGC FGCT GCT
512.0 512.0 0.0 96.0 173568.0 128136.7 349696.0 21416.8 43392.0 40499.1 5760.0 5264.3 79 0.158 2 0.057 0.215使用jvisualvm也可以看到堆內存使用情況非常穩定，流式查詢在內存使用和查詢效率上做了很好的平衡：
四、流式查詢使用注意事項
流式查詢在使用過程中當前客戶端連接會持有本次查詢的ResultSet，如果沒有沒有將這個ResultSet關閉將會影響其他查詢使用。
在數據量較大的情況下，流式查詢會長時間持有當前數據庫連接，所以還是可能存在網絡擁塞的風險。
