последовательность
В этой статье в основном изучается fetchSize оператора jdbc.
fetchSize
Здесь в качестве примера мы возьмем драйвер postgres jdbc, в основном потому, что драйвер postgres jdbc имеет открытый исходный код, а имена относительно стандартизированы. Глядя на oracle jdbc раньше, из-за отсутствия исходного кода многие имена переменных, такие как var1, var2 и т. д., декомпилируются, что очень неясно.
По умолчанию драйвер pgjdbc извлекает все наборы результатов сразу, то есть при выполнении запроса. Для запросов с большими объемами данных очень легко вызвать OOM. В этом сценарии необходимо установить fetchSize.При выполнении запроса сначала возвращается первый пакет данных, а затем после завершения следующего пакета данных извлекается следующий пакет данных.
Но к этому есть несколько требований:
- База данных должна использовать протокол V3, т.е. pg7.4+
- AutoCommit соединения должен быть ложным, потому что, если autoCommit включен, курсор будет закрыт после завершения запроса, поэтому его нельзя будет получить в следующий раз. Кроме того, ResultSet должен иметь тип ResultSet.TYPE_FORWARD_ONLY, который используется по умолчанию. То есть нет возможности прокрутить назад.
- Оператор запроса должен быть одним запросом, а не несколькими запросами, состоящими из точек с запятой.
пример кода
@Test
public void testReadTimeout() throws SQLException {
Connection connection = dataSource.getConnection();
//https://jdbc.postgresql.org/documentation/head/query.html
connection.setAutoCommit(false); //NOTE 为了设置fetchSize,必须设置为false
String sql = "select * from demo_table";
PreparedStatement pstmt;
try {
pstmt = (PreparedStatement)connection.prepareStatement(sql);
pstmt.setFetchSize(50);
System.out.println("ps.getQueryTimeout():" + pstmt.getQueryTimeout());
System.out.println("ps.getFetchSize():" + pstmt.getFetchSize());
System.out.println("ps.getFetchDirection():" + pstmt.getFetchDirection());
System.out.println("ps.getMaxFieldSize():" + pstmt.getMaxFieldSize());
ResultSet rs = pstmt.executeQuery();
//NOTE 这里返回了就代表statement执行完成,默认返回fetchSize的数据
int col = rs.getMetaData().getColumnCount();
System.out.println("============================");
while (rs.next()) {
for (int i = 1; i <= col; i++) {
System.out.print(rs.getObject(i));
}
System.out.println("");
}
System.out.println("============================");
} catch (SQLException e) {
e.printStackTrace();
} finally {
//close resources
}
}
Анализ исходного кода
postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgPreparedStatement.java
/*
* A Prepared SQL query is executed and its ResultSet is returned
*
* @return a ResultSet that contains the data produced by the * query - never null
*
* @exception SQLException if a database access error occurs
*/
public java.sql.ResultSet executeQuery() throws SQLException {
if (!executeWithFlags(0)) {
throw new PSQLException(GT.tr("No results were returned by the query."), PSQLState.NO_DATA);
}
if (result.getNext() != null) {
throw new PSQLException(GT.tr("Multiple ResultSets were returned by the query."),
PSQLState.TOO_MANY_RESULTS);
}
return result.getResultSet();
}
executeQuery сначала вызывает метод executeWithFlags. Исходный код записывается непосредственно в if. Это не рекомендуемый метод, поскольку его легче игнорировать, если он помещен в if.
- executeWithFlags
public boolean executeWithFlags(int flags) throws SQLException {
try {
checkClosed();
if (connection.getPreferQueryMode() == PreferQueryMode.SIMPLE) {
flags |= QueryExecutor.QUERY_EXECUTE_AS_SIMPLE;
}
execute(preparedQuery, preparedParameters, flags);
return (result != null && result.getResultSet() != null);
} finally {
defaultTimeZone = null;
}
}
protected final void execute(CachedQuery cachedQuery, ParameterList queryParameters, int flags)
throws SQLException {
try {
executeInternal(cachedQuery, queryParameters, flags);
} catch (SQLException e) {
// Don't retry composite queries as it might get partially executed
if (cachedQuery.query.getSubqueries() != null
|| !connection.getQueryExecutor().willHealOnRetry(e)) {
throw e;
}
cachedQuery.query.close();
// Execute the query one more time
executeInternal(cachedQuery, queryParameters, flags);
}
}
Здесь снова вызывается метод execute, и вызывается executeInternal
executeInternal
postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgPreparedStatement.java
private void executeInternal(CachedQuery cachedQuery, ParameterList queryParameters, int flags)
throws SQLException {
closeForNextExecution();
// Enable cursor-based resultset if possible.
if (fetchSize > 0 && !wantsScrollableResultSet() && !connection.getAutoCommit()
&& !wantsHoldableResultSet()) {
flags |= QueryExecutor.QUERY_FORWARD_CURSOR;
}
if (wantsGeneratedKeysOnce || wantsGeneratedKeysAlways) {
flags |= QueryExecutor.QUERY_BOTH_ROWS_AND_STATUS;
// If the no results flag is set (from executeUpdate)
// clear it so we get the generated keys results.
//
if ((flags & QueryExecutor.QUERY_NO_RESULTS) != 0) {
flags &= ~(QueryExecutor.QUERY_NO_RESULTS);
}
}
if (isOneShotQuery(cachedQuery)) {
flags |= QueryExecutor.QUERY_ONESHOT;
}
// Only use named statements after we hit the threshold. Note that only
// named statements can be transferred in binary format.
if (connection.getAutoCommit()) {
flags |= QueryExecutor.QUERY_SUPPRESS_BEGIN;
}
// updateable result sets do not yet support binary updates
if (concurrency != ResultSet.CONCUR_READ_ONLY) {
flags |= QueryExecutor.QUERY_NO_BINARY_TRANSFER;
}
Query queryToExecute = cachedQuery.query;
if (queryToExecute.isEmpty()) {
flags |= QueryExecutor.QUERY_SUPPRESS_BEGIN;
}
if (!queryToExecute.isStatementDescribed() && forceBinaryTransfers
&& (flags & QueryExecutor.QUERY_EXECUTE_AS_SIMPLE) == 0) {
// Simple 'Q' execution does not need to know parameter types
// When binaryTransfer is forced, then we need to know resulting parameter and column types,
// thus sending a describe request.
int flags2 = flags | QueryExecutor.QUERY_DESCRIBE_ONLY;
StatementResultHandler handler2 = new StatementResultHandler();
connection.getQueryExecutor().execute(queryToExecute, queryParameters, handler2, 0, 0,
flags2);
ResultWrapper result2 = handler2.getResults();
if (result2 != null) {
result2.getResultSet().close();
}
}
StatementResultHandler handler = new StatementResultHandler();
result = null;
try {
startTimer();
connection.getQueryExecutor().execute(queryToExecute, queryParameters, handler, maxrows,
fetchSize, flags);
} finally {
killTimerTask();
}
result = firstUnclosedResult = handler.getResults();
if (wantsGeneratedKeysOnce || wantsGeneratedKeysAlways) {
generatedKeys = result;
result = result.getNext();
if (wantsGeneratedKeysOnce) {
wantsGeneratedKeysOnce = false;
}
}
}
В основном смотреть это
connection.getQueryExecutor().execute(queryToExecute, queryParameters, handler, maxrows,
fetchSize, flags);
Извлеките результат указанного размера, передав fetchSize
Наконец, вызовите методы sendExecute и processResults для извлечения данных. postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/core/v3/QueryExecutorImpl.java
private void sendExecute(SimpleQuery query, Portal portal, int limit) throws IOException {
//
// Send Execute.
//
if (logger.logDebug()) {
logger.debug(" FE=> Execute(portal=" + portal + ",limit=" + limit + ")");
}
byte[] encodedPortalName = (portal == null ? null : portal.getEncodedPortalName());
int encodedSize = (encodedPortalName == null ? 0 : encodedPortalName.length);
// Total size = 4 (size field) + 1 + N (source portal) + 4 (max rows)
pgStream.sendChar('E'); // Execute
pgStream.sendInteger4(4 + 1 + encodedSize + 4); // message size
if (encodedPortalName != null) {
pgStream.send(encodedPortalName); // portal name
}
pgStream.sendChar(0); // portal name terminator
pgStream.sendInteger4(limit); // row limit
pendingExecuteQueue.add(new ExecuteRequest(query, portal, false));
}
protected void processResults(ResultHandler handler, int flags) throws IOException {
boolean noResults = (flags & QueryExecutor.QUERY_NO_RESULTS) != 0;
boolean bothRowsAndStatus = (flags & QueryExecutor.QUERY_BOTH_ROWS_AND_STATUS) != 0;
List<byte[][]> tuples = null;
int c;
boolean endQuery = false;
// At the end of a command execution we have the CommandComplete
// message to tell us we're done, but with a describeOnly command
// we have no real flag to let us know we're done. We've got to
// look for the next RowDescription or NoData message and return
// from there.
boolean doneAfterRowDescNoData = false;
while (!endQuery) {
c = pgStream.receiveChar();
switch (c) {
case 'A': // Asynchronous Notify
receiveAsyncNotify();
break;
case '1': // Parse Complete (response to Parse)
pgStream.receiveInteger4(); // len, discarded
SimpleQuery parsedQuery = pendingParseQueue.removeFirst();
String parsedStatementName = parsedQuery.getStatementName();
//...
}
}
}
next
postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgResultSet.java
public boolean next() throws SQLException {
checkClosed();
if (onInsertRow) {
throw new PSQLException(GT.tr("Can''t use relative move methods while on the insert row."),
PSQLState.INVALID_CURSOR_STATE);
}
if (current_row + 1 >= rows.size()) {
if (cursor == null || (maxRows > 0 && row_offset + rows.size() >= maxRows)) {
current_row = rows.size();
this_row = null;
rowBuffer = null;
return false; // End of the resultset.
}
// Ask for some more data.
row_offset += rows.size(); // We are discarding some data.
int fetchRows = fetchSize;
if (maxRows != 0) {
if (fetchRows == 0 || row_offset + fetchRows > maxRows) {
// Fetch would exceed maxRows, limit it.
fetchRows = maxRows - row_offset;
}
}
// Execute the fetch and update this resultset.
connection.getQueryExecutor().fetch(cursor, new CursorResultHandler(), fetchRows);
current_row = 0;
// Test the new rows array.
if (rows.isEmpty()) {
this_row = null;
rowBuffer = null;
return false;
}
} else {
current_row++;
}
initRowBuffer();
return true;
}
Как видно из следующего метода, сначала определите, меньше ли current_row + 1, чем rows.size(), если меньше, то current_row++, в противном случае это означает, что этот пакет данных fetchSize израсходован, и вам нужно чтобы решить, следует ли закончить или извлечь следующий пакет данных, обновите current_row после
connection.getQueryExecutor().fetch(cursor, new CursorResultHandler(), fetchRows);
Этот метод извлекает следующий пакет данных для количества fetchRows.
- initRowBuffer
private void initRowBuffer() {
this_row = rows.get(current_row);
// We only need a copy of the current row if we're going to
// modify it via an updatable resultset.
if (resultsetconcurrency == ResultSet.CONCUR_UPDATABLE) {
rowBuffer = new byte[this_row.length][];
System.arraycopy(this_row, 0, rowBuffer, 0, this_row.length);
} else {
rowBuffer = null;
}
}
Это нужно для того, чтобы поместить строку данных, которые будут использоваться, в rowBuffer после следующего перемещения.
резюме
Для сценариев с большим объемом данных запроса очень необходимо установить fetchSize, иначе легко OOM для полного вытягивания, но при использовании fetchSize требуется, чтобы данные могли быть обработаны вовремя при обходе результирующего набора, а не чем вернуться после сбора всех данных.