专栏首页码匠的流水账聊聊pg jdbc的queryTimeout及next方法

聊聊pg jdbc的queryTimeout及next方法

本文主要介绍一下pg jdbc statement的queryTimeout及resultSet的next方法

实例程序

    @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.setQueryTimeout(1); //NOTE 设置Statement执行完成的超时时间,前提是socket的timeout比这个大
            pstmt.setFetchSize(5000); //NOTE 这样设置为了模拟query timeout的异常
            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执行完成的超时时间,前提是socket的timeout比这个大
            //NOTE 这里返回了就代表statement执行完成,默认返回fetchSize的数据
            int col = rs.getMetaData().getColumnCount();
            System.out.println("============================");
            while (rs.next()) { //NOTE 这个的timeout由socket的超时时间设置,oracle.jdbc.ReadTimeout=60000
                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
        }
    }

PgStatement

ostgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgStatement.java

executeInternal()

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;
      }
    }

  }

注意,这里在执行前后分别调用了startTimer()和killTimerTask()

startTimer()

private void startTimer() {
    /*
     * there shouldn't be any previous timer active, but better safe than sorry.
     */
    cleanupTimer();

    STATE_UPDATER.set(this, StatementCancelState.IN_QUERY);

    if (timeout == 0) {
      return;
    }

    TimerTask cancelTask = new TimerTask() {
      public void run() {
        try {
          if (!CANCEL_TIMER_UPDATER.compareAndSet(PgStatement.this, this, null)) {
            // Nothing to do here, statement has already finished and cleared
            // cancelTimerTask reference
            return;
          }
          PgStatement.this.cancel();
        } catch (SQLException e) {
        }
      }
    };

    CANCEL_TIMER_UPDATER.set(this, cancelTask);
    connection.addTimerTask(cancelTask, timeout);
  }
  • startTimer调用了cleanupTimer()
  • cancelTask调用的是PgStatement.this.cancel()
  • 最后调用connection.addTimerTask添加定时任务

cleanupTimer()

/**
   * Clears {@link #cancelTimerTask} if any. Returns true if and only if "cancel" timer task would
   * never invoke {@link #cancel()}.
   */
  private boolean cleanupTimer() {
    TimerTask timerTask = CANCEL_TIMER_UPDATER.get(this);
    if (timerTask == null) {
      // If timeout is zero, then timer task did not exist, so we safely report "all clear"
      return timeout == 0;
    }
    if (!CANCEL_TIMER_UPDATER.compareAndSet(this, timerTask, null)) {
      // Failed to update reference -> timer has just fired, so we must wait for the query state to
      // become "cancelling".
      return false;
    }
    timerTask.cancel();
    connection.purgeTimerTasks();
    // All clear
    return true;
  }

注意这里更新statement状态之后,调用task的cancel,以及connection.purgeTimerTasks()

cancel()

public void cancel() throws SQLException {
    if (!STATE_UPDATER.compareAndSet(this, StatementCancelState.IN_QUERY, StatementCancelState.CANCELING)) {
      // Not in query, there's nothing to cancel
      return;
    }
    try {
      // Synchronize on connection to avoid spinning in killTimerTask
      synchronized (connection) {
        connection.cancelQuery();
      }
    } finally {
      STATE_UPDATER.set(this, StatementCancelState.CANCELLED);
      synchronized (connection) {
        connection.notifyAll(); // wake-up killTimerTask
      }
    }
  }

executeQuery超时了则直接调用connection.cancelQuery()

  public void cancelQuery() throws SQLException {
    checkClosed();
    queryExecutor.sendQueryCancel();
  }

postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/core/QueryExecutorBase.java

public void sendQueryCancel() throws SQLException {
    if (cancelPid <= 0) {
      return;
    }

    PGStream cancelStream = null;

    // Now we need to construct and send a cancel packet
    try {
      if (logger.logDebug()) {
        logger.debug(" FE=> CancelRequest(pid=" + cancelPid + ",ckey=" + cancelKey + ")");
      }

      cancelStream =
          new PGStream(pgStream.getSocketFactory(), pgStream.getHostSpec(), cancelSignalTimeout);
      if (cancelSignalTimeout > 0) {
        cancelStream.getSocket().setSoTimeout(cancelSignalTimeout);
      }
      cancelStream.sendInteger4(16);
      cancelStream.sendInteger2(1234);
      cancelStream.sendInteger2(5678);
      cancelStream.sendInteger4(cancelPid);
      cancelStream.sendInteger4(cancelKey);
      cancelStream.flush();
      cancelStream.receiveEOF();
    } catch (IOException e) {
      // Safe to ignore.
      if (logger.logDebug()) {
        logger.debug("Ignoring exception on cancel request:", e);
      }
    } finally {
      if (cancelStream != null) {
        try {
          cancelStream.close();
        } catch (IOException e) {
          // Ignored.
        }
      }
    }
  }

向数据库server发送cancel指令

killTimerTask()

private void killTimerTask() {
    boolean timerTaskIsClear = cleanupTimer();
    // The order is important here: in case we need to wait for the cancel task, the state must be
    // kept StatementCancelState.IN_QUERY, so cancelTask would be able to cancel the query.
    // It is believed that this case is very rare, so "additional cancel and wait below" would not
    // harm it.
    if (timerTaskIsClear && STATE_UPDATER.compareAndSet(this, StatementCancelState.IN_QUERY, StatementCancelState.IDLE)) {
      return;
    }

    // Being here means someone managed to call .cancel() and our connection did not receive
    // "timeout error"
    // We wait till state becomes "cancelled"
    boolean interrupted = false;
    while (!STATE_UPDATER.compareAndSet(this, StatementCancelState.CANCELLED, StatementCancelState.IDLE)) {
      synchronized (connection) {
        try {
          // Note: wait timeout here is irrelevant since synchronized(connection) would block until
          // .cancel finishes
          connection.wait(10);
        } catch (InterruptedException e) { // NOSONAR
          // Either re-interrupt this method or rethrow the "InterruptedException"
          interrupted = true;
        }
      }
    }
    if (interrupted) {
      Thread.currentThread().interrupt();
    }
  }

这里先调用cleanupTimer,然后更新statement的状态

PgConnection

postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgConnection.java

getTimer()

private synchronized Timer getTimer() {
    if (cancelTimer == null) {
      cancelTimer = Driver.getSharedTimer().getTimer();
    }
    return cancelTimer;
  }

这里创建或获取一个timer

addTimerTask()

  public void addTimerTask(TimerTask timerTask, long milliSeconds) {
    Timer timer = getTimer();
    timer.schedule(timerTask, milliSeconds);
  }

这个添加timerTask就是直接调度了

purgeTimerTasks()

postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgConnection.java

public void purgeTimerTasks() {
    Timer timer = cancelTimer;
    if (timer != null) {
      timer.purge();
    }
  }

在cleanupTimer中被调用,用来清理已经被cancel掉的timer task

PgResultSet

postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/jdbc/PgResultSet.java

next()

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;
  }

这里的fetch没有像executeQuery那样加timer

postgresql-9.4.1212.jre7-sources.jar!/org/postgresql/core/v3/QueryExecutorImpl.java

public synchronized void fetch(ResultCursor cursor, ResultHandler handler, int fetchSize)
      throws SQLException {
    waitOnLock();
    final Portal portal = (Portal) cursor;

    // Insert a ResultHandler that turns bare command statuses into empty datasets
    // (if the fetch returns no rows, we see just a CommandStatus..)
    final ResultHandler delegateHandler = handler;
    handler = new ResultHandlerDelegate(delegateHandler) {
      public void handleCommandStatus(String status, int updateCount, long insertOID) {
        handleResultRows(portal.getQuery(), null, new ArrayList<byte[][]>(), null);
      }
    };

    // Now actually run it.

    try {
      processDeadParsedQueries();
      processDeadPortals();

      sendExecute(portal.getQuery(), portal, fetchSize);
      sendSync();

      processResults(handler, 0);
      estimatedReceiveBufferBytes = 0;
    } catch (IOException e) {
      abort();
      handler.handleError(
          new PSQLException(GT.tr("An I/O error occurred while sending to the backend."),
              PSQLState.CONNECTION_FAILURE, e));
    }

    handler.handleCompletion();
  }

小结

  • queryTimeout是采用添加timer来控制,如果请求过多,可能会造成timer过多 timeout时间不宜过长,不过正常执行完sql,会调用killTimerTask()方,里头会先cleanupTimer,取消timerTask,然后调用purgeTimerTasks()清理cancel掉的task,避免timeout时间过长导致task堆积最后内存溢出
  • 超时之后会timer task会向数据库server发送cancel query指令
  • 发送完cancel query指令之后,client端的查询按预期应该抛出SQLException(这里头的机制有待深入研究,可能是server端返回timeout error)
  • executeQuery方法默认会拉取fetchSize的数据并返回
  • next()方法根据需要再去fetch,这个fetch方法就没有timer来限制时间了,但是最底层应该是受socketTimeout限制

doc

  • MySQL JDBC的queryTimeout的一个坑

本文分享自微信公众号 - 码匠的流水账(geek_luandun),作者:code-craft

原文出处及转载信息见文内详细说明,如有侵权,请联系 yunjia_community@tencent.com 删除。

原始发表时间:2018-01-24

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • 聊聊artemis的handleConnectionFailure

    activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/ar...

    codecraft
  • 聊聊artemis的handleConnectionFailure

    activemq-artemis-2.11.0/artemis-core-client/src/main/java/org/apache/activemq/ar...

    codecraft
  • 聊聊skywalking的MemoryProvider

    skywalking-6.6.0/apm-sniffer/apm-agent-core/src/main/java/org/apache/skywalking/...

    codecraft
  • dubbo源码学习笔记----结合Spring

    dubbo结合spring public class ServiceBean<T> extends ServiceConfig<T> implements In...

    春哥大魔王
  • 分析3000份技术面试数据:这几大指标比你毕业于哪所学校更要紧

    大数据文摘
  • 阿里,网易游戏,腾讯,smartx的offer面试分享

    从今年的3月14日阿里的电话面试开始,到现在4月16日在西安悦豪酒店进行的腾讯HR面到现在一个多月了,中间先后收到了阿里,网易游戏,腾讯和smartx的offe...

    Java架构技术
  • GitHub 上四万 Star 大佬的求职回忆

    原文链接:https://www.nowcoder.com/discuss/137593

    用户2802329
  • display:none和visibility:hidden的区别

    pitaojin
  • Python爬虫:Scrapy 框架快速入门及实战演练

    今天来介绍一下 Python 的一个爬虫框架Scrapy ,类似的还有 Portia Crawley。

    wsuo
  • 程序员才看得明白的面试圣经

    除了直接申请面试以外,一般说来,还有两种途径来获得面试的机会:由现在的雇主推荐,或者通过LinkedIn。虽然前者会快一些、更尊敬一些,但后者很可能是大部分应聘...

    哲洛不闹

扫码关注云+社区

领取腾讯云代金券