很久沒(méi)寫(xiě)開(kāi)篇了，針對(duì)大數(shù)據(jù)采集分析和導(dǎo)出等功能，我們必須在內(nèi)存和性能上做好折中，這其中筆者最?lèi)?ài)的就是流式查詢(xún)，而本文將基于多個(gè)角度針對(duì)流式查詢(xún)這技能進(jìn)行深入的分析和演示，希望對(duì)你有幫助。

一、詳解流式查詢(xún)

1. 關(guān)于IOPS和數(shù)據(jù)吞吐量

為了保證后文講解的流暢，我們這里對(duì)幾個(gè)比較重要的性能指標(biāo)進(jìn)行簡(jiǎn)單的科普，對(duì)于服務(wù)器系統(tǒng)層面，IOPS(Input/Output Per Second)磁盤(pán)每秒的讀寫(xiě)次數(shù)，一般以每秒輸入輸出量為單位進(jìn)行衡量。而吞吐量更多的是反應(yīng)的是每秒處理的IO請(qǐng)求，兩者關(guān)系我們可以通過(guò)如下兩個(gè)場(chǎng)景了解一下差異：

• 假設(shè)我們讀取1000個(gè)1kb的數(shù)據(jù)，耗時(shí)10s，那么這個(gè)服務(wù)器的數(shù)據(jù)吞吐量100KB/s，IOPS就是100，這種場(chǎng)景更追求IOPS。
• 假設(shè)我們只有1個(gè)請(qǐng)求去讀取10M的文件，耗時(shí)0.2s，那么這個(gè)服務(wù)器的數(shù)據(jù)吞吐量就是50MB/s，IOPS為5，這個(gè)服務(wù)器就更偏向于吞吐量。

2. MySQL常見(jiàn)的幾種查詢(xún)

日常針對(duì)大表數(shù)據(jù)采集導(dǎo)出的功能，我們一般會(huì)采用一下幾種方案：

• 一次性全量導(dǎo)出
• 使用分頁(yè)查詢(xún)
• 使用游標(biāo)查詢(xún)
• 流式查詢(xún)

(1) 全量導(dǎo)出

我們先來(lái)說(shuō)說(shuō)全量查詢(xún)，這種方案本質(zhì)原理就是一次性將結(jié)果集從MySQL服務(wù)端寫(xiě)到客戶(hù)端程序上，針對(duì)大表數(shù)據(jù)檢索，如果我們的程序沒(méi)有足夠的堆內(nèi)存空間，存在內(nèi)存溢出的風(fēng)險(xiǎn)：

(2) 分頁(yè)查詢(xún)

為了解決OOM問(wèn)題，我們會(huì)考慮通過(guò)分頁(yè)查詢(xún)的方式，通過(guò)分批處理完成批量數(shù)據(jù)檢索導(dǎo)出的工作，這種方式雖然很好的節(jié)約了堆內(nèi)存空間，但這種方案在代碼實(shí)現(xiàn)層面就已經(jīng)非常復(fù)雜了，開(kāi)發(fā)者必須考慮：

• 分頁(yè)計(jì)算(這一步就涉及數(shù)據(jù)掃描，開(kāi)銷(xiāo)大)
• 基于分頁(yè)評(píng)估每次分頁(yè)大小
• 基于頁(yè)數(shù)進(jìn)行循環(huán)查詢(xún)
• 查詢(xún)SQL需要針對(duì)深分頁(yè)問(wèn)題進(jìn)行優(yōu)化

這種方案相較于前者雖然節(jié)省了堆內(nèi)存空間且可以一定程度上避免頻繁的Full GC，對(duì)于開(kāi)發(fā)者整體素質(zhì)要求較高，并且這種方案在性能表現(xiàn)上也不是很出色：

(3) 游標(biāo)查詢(xún)

所以為了避免在開(kāi)發(fā)層面進(jìn)行手動(dòng)分頁(yè)實(shí)現(xiàn)的復(fù)雜度，我們就想到通過(guò)游標(biāo)法進(jìn)行查詢(xún)，游標(biāo)也就是cursor，這種查詢(xún)方式要求客戶(hù)端一次性指明fetchSize，然后服務(wù)端每次都基于給定的fetchSize將數(shù)據(jù)寫(xiě)給客戶(hù)端，直到客戶(hù)端將所有數(shù)據(jù)都處理完成。

需要了解的是游標(biāo)查詢(xún)這種方案考慮到客戶(hù)端未知的處理效率，為保證服務(wù)端能夠一次性將fetch的數(shù)據(jù)寫(xiě)回到客戶(hù)端，MySQL服務(wù)端會(huì)為了這個(gè)查詢(xún)建立一個(gè)臨時(shí)空間來(lái)緩存數(shù)據(jù)，在極端情況下因?yàn)檫@些問(wèn)題：

• IOPS飆升
• 磁盤(pán)空間飆升(因?yàn)榕R時(shí)空間無(wú)法在緩存中容納，寫(xiě)入到文件中)
• fetch設(shè)置過(guò)大，SQL查詢(xún)經(jīng)常處于阻塞等待IO數(shù)據(jù)的情況

(4) 流式查詢(xún)

最后我們就來(lái)說(shuō)說(shuō)本文的重點(diǎn)——流式查詢(xún)，當(dāng)客戶(hù)端向服務(wù)端發(fā)送SQL請(qǐng)求后，流式查詢(xún)會(huì)得到一個(gè)迭代器，客戶(hù)端不斷通過(guò)ResultSet.next()獲取下一條數(shù)據(jù)，服務(wù)端會(huì)按照客戶(hù)端接受速率并基于迭代器的偏移量逐步寫(xiě)入到網(wǎng)絡(luò)buffer中讓客戶(hù)端讀取，這種方式很好的解決游標(biāo)查詢(xún)逐批次緩存的問(wèn)題。 但需要注意的是這種方案和上述游標(biāo)查詢(xún)一樣，會(huì)因?yàn)閿?shù)據(jù)量的問(wèn)題，使得連接長(zhǎng)時(shí)間被當(dāng)前線(xiàn)程持有：

二、流式查詢(xún)使用示例

接下來(lái)筆者就以常見(jiàn)的ORM框架Mybatis演示一下如何使用流式查詢(xún)，假設(shè)我們需要查詢(xún)一張user表，對(duì)應(yīng)的我們基于Options注解給出當(dāng)前這個(gè)查詢(xún)信息告知查詢(xún)r(jià)esultSetType 為只讀，并且指明fetchSize 為MIN_VALUE。同時(shí)，看到筆者在方法上給出了一個(gè)ResultHandler，這個(gè)處理用于處理流式查詢(xún)響應(yīng)結(jié)果后的回調(diào)處理：

@Select("select * from user  ")
    @Options(resultSetType = ResultSetType.FORWARD_ONLY, fetchSize = Integer.MIN_VALUE)
    @ResultType(User.class)
    void selectListByStream(ResultHandler<User> handler);

需要補(bǔ)充說(shuō)明的是上述三個(gè)配置都必須明確按照要求進(jìn)行配置，原因是在mybatis在執(zhí)行SQL查詢(xún)時(shí)，StatementImpl會(huì)通過(guò)createStreamingResultSet判斷這三個(gè)參數(shù)是否符合要求，只有明確符合要求返回true，后續(xù)的結(jié)果集才能被創(chuàng)建為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));
    }

對(duì)應(yīng)的我們也給出最后的使用示例，這里筆者用lambda精簡(jiǎn)了一下ResultHandler的聲明，每當(dāng)我們收到流式響應(yīng)數(shù)據(jù)后，直接獲取user并自增一下原子類(lèi)：

userMapper.selectListByStream(resultContext -> {
                User user = resultContext.getResultObject();
               //執(zhí)行業(yè)務(wù)操作......           

            });

三、性能壓測(cè)

簡(jiǎn)單介紹了一下流式查詢(xún)的基礎(chǔ)配置和使用之后，我們不妨針對(duì)上述方案進(jìn)行性能和內(nèi)存使用情況壓測(cè)，首先筆者已經(jīng)準(zhǔn)備了100w條數(shù)據(jù)并將堆內(nèi)存設(shè)置為512M：

-Xmx512m -Xms512m

我們先給出一個(gè)基于全量查詢(xún)的導(dǎo)出寫(xiě)入到本地txt文件：

try (BufferedWriter outputStream = FileUtil.getWriter("F://tmp/userData.txt", Charset.defaultCharset(), false)) {
            //全量查詢(xún)
            List<User> userList = userMapper.selectList(Wrappers.emptyWrapper());
            //遍歷寫(xiě)入本地文件
            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數(shù)據(jù)導(dǎo)出跑了大約20s:

save count:1000000 cost:20382ms

我們通過(guò)jstat指令查看堆內(nèi)存使用情況，觸發(fā)了6次full gc，整體回收花費(fèi)了6s，性能表現(xiàn)非常差勁：

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

然后就是分頁(yè)查詢(xún)，可以看到筆者這里并沒(méi)有針對(duì)深分頁(yè)問(wèn)題進(jìn)行優(yōu)化，明確頁(yè)數(shù)和分批數(shù)后直接進(jìn)行分批查詢(xún)導(dǎo)出了：

//定位總數(shù)
        Long c = userMapper.selectCount(Wrappers.emptyWrapper());
        //計(jì)算頁(yè)數(shù)
        int size = 1_0000;
        long pageSize = c % size == 0 ? c / size : c / size + 1;
        //分頁(yè)查詢(xún)
        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);
        }

可以看到這段代碼內(nèi)存使用情況比較穩(wěn)定，但是耗時(shí)大約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

最后我們給出流式查詢(xún)的代碼拉取數(shù)據(jù)并寫(xiě)入本地文件的示例：

try (BufferedWriter outputStream = FileUtil.getWriter("F://tmp/userData.txt", Charset.defaultCharset(), false)) {
            //流式查詢(xún)
            userMapper.selectListByStream(res -> {
                //定位對(duì)象
                User user = res.getResultObject();
                try {
                    //寫(xiě)入文件
                    outputStream.write(JSONUtil.toJsonStr(user) + "\r\n");
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                count.incrementAndGet();

            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

最終整體耗時(shí)14s，使用jstat查看gc情況也是非常穩(wěn)定：

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也可以看到堆內(nèi)存使用情況非常穩(wěn)定，流式查詢(xún)?cè)趦?nèi)存使用和查詢(xún)效率上做了很好的平衡：

四、流式查詢(xún)使用注意事項(xiàng)

流式查詢(xún)?cè)谑褂眠^(guò)程中當(dāng)前客戶(hù)端連接會(huì)持有本次查詢(xún)的ResultSet，如果沒(méi)有沒(méi)有將這個(gè)ResultSet關(guān)閉將會(huì)影響其他查詢(xún)使用。

在數(shù)據(jù)量較大的情況下，流式查詢(xún)會(huì)長(zhǎng)時(shí)間持有當(dāng)前數(shù)據(jù)庫(kù)連接，所以還是可能存在網(wǎng)絡(luò)擁塞的風(fēng)險(xiǎn)。