JDBC查詢

使用JDBC向mysql發(fā)送查詢時(shí)，有三種方式：

1. 常規(guī)查詢：JDBC驅(qū)動(dòng)會(huì)阻塞的一次性讀取全部查詢的數(shù)據(jù)到 JVM 內(nèi)存中，或者分頁(yè)讀取
2. 流式查詢：每次執(zhí)行rs.next時(shí)會(huì)判斷數(shù)據(jù)是否需要從mysql服務(wù)器獲取，如果需要觸發(fā)讀取一批數(shù)據(jù)（可能n行）加載到 JVM 內(nèi)存進(jìn)行業(yè)務(wù)處理
3. 游標(biāo)查詢：通過 fetchSize 參數(shù)，控制每次從mysql服務(wù)器一次讀取多少行數(shù)據(jù)。

1、常規(guī)查詢

public static void normalQuery() throws SQLException {
    Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3307/test?useSSL=false", "root", "123456");
    PreparedStatement statement = connection.prepareStatement(sql);
    //statement.setFetchSize(100); //不起作用
    ResultSet resultSet = statement.executeQuery();
    while(resultSet.next()){
        System.out.println(resultSet.getString(2));
    }
    resultSet.close();
    statement.close();
    connection.close();
}

1）說明：

第四行設(shè)置featchSize不起作用。第五行statement.executeQuery()執(zhí)行查詢會(huì)阻塞，因?yàn)樾枰鹊剿袛?shù)據(jù)返回并放到內(nèi)存中；接下來每次執(zhí)行resultSet.next()方法會(huì)從內(nèi)存中獲取數(shù)據(jù)。

2）將jvm內(nèi)存設(shè)置較?。?Xms16m -Xmx16m），對(duì)于大數(shù)據(jù)的查詢會(huì)產(chǎn)生OOM：

為了避免OOM，通常我們會(huì)使用分頁(yè)查詢，或者下面的兩種方式。

2、流式查詢

public static void streamQuery() throws Exception { 
    Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3307/test?useSSL=false", "root", "123456");
    PreparedStatement statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);    
    statement.setFetchSize(Integer.MIN_VALUE); 
    //或者通過 com.mysql.jdbc.StatementImpl
    ((StatementImpl) statement).enableStreamingResults();
    ResultSet rs = statement.executeQuery();
    while (rs.next()) {
        System.out.println(rs.getString(2));
    }
    rs.close();
    statement.close();
    connection.close();
}

2.1 流式查詢的條件：

隨著大數(shù)據(jù)的到來，對(duì)于百萬、千萬的數(shù)據(jù)使用流式查詢可以有效避免OOM。

在執(zhí)行statement.executeQuery()時(shí)不會(huì)從TCP響應(yīng)流中讀取完所有數(shù)據(jù)，當(dāng)下面執(zhí)行rs.next()時(shí)會(huì)按照需要從TCP響應(yīng)流中讀取部分?jǐn)?shù)據(jù)。

• 創(chuàng)建Statement的時(shí)候需要制定ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY
• 設(shè)置fetchSize位Integer.MIN_VALUE

或者通過com.mysql.jdbc.StatementImpl的enableStreamingResults()方法設(shè)置。

二者是一致的。看mysql的jdbc（com.mysql.jdbc.StatementImpl）

源碼:

2.2 流式查詢?cè)恚?/h3>

1）基本概念

我們要知道jdbc客戶端和mysql服務(wù)器之間是通過TCP建立的通信，使用mysql協(xié)議進(jìn)行傳輸數(shù)據(jù)。

首先聲明一個(gè)概念：在三次握手建立了TCP連接后，就可以在這個(gè)通道上進(jìn)行通信了，直到關(guān)閉該連接。

在 TCP 中發(fā)送端和接收端**可以是客戶端/服務(wù)端，也可以是服務(wù)器/客戶端**，通信的雙方在任意時(shí)刻既可以是接收數(shù)據(jù)也可以是發(fā)送數(shù)據(jù)（全雙工）。

在通信中，收發(fā)雙方都不保持記錄的邊界，所以需要按照一定的協(xié)議進(jìn)行表示。在mysql中會(huì)按照mysql協(xié)議來進(jìn)行交互。

有了上面的概念，我們重新來定義這兩種查詢：

在執(zhí)行st.executeQuery()時(shí)，jdbc驅(qū)動(dòng)會(huì)通過connection對(duì)象和mysql服務(wù)器建立TCP連接，同時(shí)在這個(gè)鏈接通道中發(fā)送sql命令，并接受返回。

二者的區(qū)別是：

• 普通查詢：也叫批量查詢，jdbc客戶端會(huì)阻塞的一次性從TCP通道中讀取完mysql服務(wù)的返回?cái)?shù)據(jù)；
• 流式查詢：分批的從TCP通道中讀取mysql服務(wù)返回的數(shù)據(jù)，每次讀取的數(shù)據(jù)量并不是一行（通常是一個(gè)package大?。琷dbc客戶端在調(diào)用rs.next()方法時(shí)會(huì)根據(jù)需要從TCP流通道中讀取部分?jǐn)?shù)據(jù)。（并不是每次讀區(qū)一行數(shù)據(jù)，網(wǎng)上說的幾乎都是錯(cuò)的?。?/li>

2）源碼查看：

從statement.executeQuery()方法跟進(jìn)去，主要的調(diào)用連如下：

protected ResultSetInternalMethods executeInternal(int maxRowsToRetrieve, Buffer sendPacket, boolean createStreamingResultSet, boolean queryIsSelectOnly,
            Field[] metadataFromCache, boolean isBatch) throws SQLException {
        synchronized (checkClosed().getConnectionMutex()) {
            MySQLConnection locallyScopedConnection = this.connection;
            rs = locallyScopedConnection.execSQL(this, null, maxRowsToRetrieve, sendPacket, this.resultSetType, this.resultSetConcurrency,
                            createStreamingResultSet, this.currentCatalog, metadataFromCache, isBatch);
            return rs;
        }
public ResultSetInternalMethods execSQL(StatementImpl callingStatement, String sql, int maxRows, Buffer packet, int resultSetType, int resultSetConcurrency,
            boolean streamResults, String catalog, Field[] cachedMetadata, boolean isBatch) throws SQLException {
        synchronized (getConnectionMutex()) {
            return this.io.sqlQueryDirect(callingStatement, null, null, packet, maxRows, resultSetType, resultSetConcurrency, streamResults, catalog,
                        cachedMetadata);
        }
}
final ResultSetInternalMethods sqlQueryDirect(StatementImpl callingStatement, String query, String characterEncoding, Buffer queryPacket, int maxRows,
            int resultSetType, int resultSetConcurrency, boolean streamResults, String catalog, Field[] cachedMetadata) throws Exception {
        Buffer resultPacket = sendCommand(MysqlDefs.QUERY, null, queryPacket, false, null, 0);
        ResultSetInternalMethods rs = readAllResults(callingStatement, maxRows, resultSetType, resultSetConcurrency, streamResults, catalog, resultPacket,
                    false, -1L, cachedMetadata);
        return rs;
}
ResultSetImpl readAllResults(StatementImpl callingStatement, int maxRows, int resultSetType, int resultSetConcurrency, boolean streamResults,
            String catalog, Buffer resultPacket, boolean isBinaryEncoded, long preSentColumnCount, Field[] metadataFromCache) throws SQLException {
        ResultSetImpl topLevelResultSet = readResultsForQueryOrUpdate(callingStatement, maxRows, resultSetType, resultSetConcurrency, streamResults, catalog,
                resultPacket, isBinaryEncoded, preSentColumnCount, metadataFromCache);
        return topLevelResultSet;
}
protected final ResultSetImpl readResultsForQueryOrUpdate(StatementImpl callingStatement, int maxRows, int resultSetType, int resultSetConcurrency,
            boolean streamResults, String catalog, Buffer resultPacket, boolean isBinaryEncoded, long preSentColumnCount, Field[] metadataFromCache) throws SQLException {
            com.mysql.jdbc.ResultSetImpl results = getResultSet(callingStatement, columnCount, maxRows, resultSetType, resultSetConcurrency, streamResults,
                    catalog, isBinaryEncoded, metadataFromCache);
            return results;
        }
}
protected ResultSetImpl getResultSet(StatementImpl callingStatement, long columnCount, int maxRows, int resultSetType, int resultSetConcurrency,
            boolean streamResults, String catalog, boolean isBinaryEncoded, Field[] metadataFromCache) throws SQLException {
        Buffer packet; // The packet from the server
        RowData rowData = null;
        if (!streamResults) {
            rowData = readSingleRowSet(columnCount, maxRows, resultSetConcurrency, isBinaryEncoded, (metadataFromCache == null) ? fields : metadataFromCache);
        } else {
            rowData = new RowDataDynamic(this, (int) columnCount, (metadataFromCache == null) ? fields : metadataFromCache, isBinaryEncoded);
            this.streamingData = rowData;
        }
        ResultSetImpl rs = buildResultSetWithRows(callingStatement, catalog, (metadataFromCache == null) ? fields : metadataFromCache, rowData, resultSetType,
                resultSetConcurrency, isBinaryEncoded);
        return rs;
}

說明：

1. sqlQueryDirect（）方法中的sendCommand會(huì)通過io發(fā)送sql命令請(qǐng)求到mysql服務(wù)器，并獲取返回流mysqlOutput
2. getResultSet（）方法會(huì)判斷是否是流式查詢還是批量查詢。MySQL驅(qū)動(dòng)會(huì)根據(jù)不同的參數(shù)設(shè)置選擇對(duì)應(yīng)的ResultSet實(shí)現(xiàn)類，分別對(duì)應(yīng)三種查詢方式：
   1. RowDataStatic 靜態(tài)結(jié)果集，默認(rèn)的查詢方式，普通查詢
   2. RowDataDynamic 動(dòng)態(tài)結(jié)果集，流式查詢
   3. RowDataCursor 游標(biāo)結(jié)果集，服務(wù)器端基于游標(biāo)查詢

看上述代碼（41行），對(duì)于批量查詢：readSingleRowSet方法會(huì)循環(huán)掉用nextRow方法獲取所有數(shù)據(jù)，然后放到j(luò)vm內(nèi)存的rows中：

對(duì)于流式查詢：直接創(chuàng)建RowDataDynamic對(duì)象返回。后面在掉用rs.next()獲取數(shù)據(jù)時(shí)會(huì)根據(jù)需要從mysqlOutput流中讀取數(shù)據(jù)。

2.3 流式查詢的坑：

public static void streamQuery2() throws Exception { 
    Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3307/test?useSSL=false", "root", "123456");
    //statement1
    PreparedStatement statement = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);    
    statement.setFetchSize(Integer.MIN_VALUE); 
    ResultSet rs = statement.executeQuery();
    if (rs.next()) {
        System.out.println(rs.getString(2));
    }
    //statement2
    PreparedStatement statement2 = connection.prepareStatement(sql, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);    
    statement2.setFetchSize(Integer.MIN_VALUE); 
    ResultSet rs2 = statement2.executeQuery();
    if (rs2.next()) {
        System.out.println(rs2.getString(2));
    }
//      rs.close();
//      statement.close();
//      connection.close();
}

執(zhí)行結(jié)果：

test1
java.sql.SQLException: Streaming result set com.mysql.jdbc.RowDataDynamic@45c8e616 is still active. No statements may be issued when any streaming result sets are open and in use on a given connection. Ensure that you have called .close() on any active streaming result sets before attempting more queries.
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:869)
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:865)
    at com.mysql.jdbc.MysqlIO.checkForOutstandingStreamingData(MysqlIO.java:3217)
    at com.mysql.jdbc.MysqlIO.sendCommand(MysqlIO.java:2453)
    at com.mysql.jdbc.MysqlIO.sqlQueryDirect(MysqlIO.java:2683)
    at com.mysql.jdbc.ConnectionImpl.execSQL(ConnectionImpl.java:2482)
    at com.mysql.jdbc.StatementImpl.executeSimpleNonQuery(StatementImpl.java:1465)
    at com.mysql.jdbc.StatementImpl.setupStreamingTimeout(StatementImpl.java:726)
    at com.mysql.jdbc.PreparedStatement.executeQuery(PreparedStatement.java:1939)
    at com.tencent.clue_disp_api.MysqlTest.streamQuery2(MysqlTest.java:79)
    at com.tencent.clue_disp_api.MysqlTest.main(MysqlTest.java:25)

MySQL Connector/J 5.1 Developer Guide中原文：

There are some caveats with this approach. You must read all of the rows in the result set (or close it) before you can issue any other queries on the connection, or an exception will be thrown. 也就是說當(dāng)通過流式查詢獲取一個(gè)ResultSet后，通過next迭代出所有元素之前或者調(diào)用close關(guān)閉它之前，不能使用同一個(gè)數(shù)據(jù)庫(kù)連接去發(fā)起另外一個(gè)查詢，否者拋出異常（第一次調(diào)用的正常，第二次的拋出異常）。

2.4 抓包驗(yàn)證：

查看3307 > 62169的包可以發(fā)現(xiàn)，ack都是1324，證明都是針對(duì)當(dāng)時(shí)sql請(qǐng)求的返回?cái)?shù)據(jù)。

3、游標(biāo)查詢

public static void cursorQuery() throws Exception {
    Connection connection = DriverManager.getConnection("jdbc:mysql://localhost:3307/test?useSSL=false&useCursorFetch=true", "root", "123456");
    ((JDBC4Connection) connection).setUseCursorFetch(true); //com.mysql.jdbc.JDBC4Connection
    Statement statement = connection.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);    
    statement.setFetchSize(2);    
    ResultSet rs = statement.executeQuery(sql);    
    while (rs.next()) {
        System.out.println(rs.getString(2));
        Thread.sleep(5000);
    }
    rs.close();
    statement.close();
    connection.close();
}

1）說明：

• 在連接參數(shù)中需要拼接useCursorFetch=true；
• 創(chuàng)建Statement時(shí)需要設(shè)置ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY
• 設(shè)置fetchSize控制每一次獲取多少條數(shù)據(jù)

2）抓包驗(yàn)證：

通過wireshark抓包，可以看到每執(zhí)行一次rs.next() 就會(huì)向mysql服務(wù)發(fā)送一個(gè)請(qǐng)求，同時(shí)mysql服務(wù)返回兩條數(shù)據(jù)：

3）游標(biāo)查詢需要注意的點(diǎn)：

由于MySQL方不知道客戶端什么時(shí)候?qū)?shù)據(jù)消費(fèi)完，而自身的對(duì)應(yīng)表可能會(huì)有DML寫入操作，此時(shí)MySQL需要建立一個(gè)臨時(shí)空間來存放需要拿走的數(shù)據(jù)。

因此對(duì)于當(dāng)你啟用useCursorFetch讀取大表的時(shí)候會(huì)看到MySQL上的幾個(gè)現(xiàn)象：

1. IOPS飆升 （IOPS (Input/Output Per Second)：磁盤每秒的讀寫次數(shù)）
2. 磁盤空間飆升
3. 客戶端JDBC發(fā)起SQL后，長(zhǎng)時(shí)間等待SQL響應(yīng)數(shù)據(jù)，這段時(shí)間就是服務(wù)端在準(zhǔn)備數(shù)據(jù)
4. 在數(shù)據(jù)準(zhǔn)備完成后，開始傳輸數(shù)據(jù)的階段，網(wǎng)絡(luò)響應(yīng)開始飆升，IOPS由“讀寫”轉(zhuǎn)變?yōu)?ldquo;讀取”。
5. CPU和內(nèi)存會(huì)有一定比例的上升

到此這篇關(guān)于Mysql中JDBC的三種查詢(普通、流式、游標(biāo))詳解的文章就介紹到這了,更多相關(guān)JDBC的三種查詢內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論