[大数据]druid源码解读--一个查询sql在druid中经历了什么？

druid–一个查询sql在druid中经历了什么？

druid的连接池配置中有PreparedStatementCache的配置，该信息解决了sql语句可以被预编译，并且保存在PreparedStatement这个对象中，而这个对象的存储就在PreparedStatementCache，对于oracle可以绕过数据库编译，有很大的提升，但是对于mysql，没有那么明显。

本文针对DruidPooledPreparedStatement类中的executeQuery方法进行解读，尝试了解一下具体如何做预处理的，sql如何执行，监控如何获取sql执行过程中的数据。DruidPooledPreparedStatement类对executeQuery方法的实现，这个方法里面最关键的是ResultSet rs = stmt.executeQuery()这句，stmt是PreparedStatementProxyImpl类的类对象。

DruidPooledPreparedStatement类图

源码分析

//构造方法 核心是：连接池和预处理的持有者
public DruidPooledPreparedStatement(DruidPooledConnection conn, PreparedStatementHolder holder) throws SQLException{
    super(conn, holder.statement);
    this.stmt = holder.statement;
    this.holder = holder;
    this.sql = holder.key.sql;
		// 配置项中是否打开属性poolPreparedStatements
    pooled = conn.getConnectionHolder().isPoolPreparedStatements();
    // Remember the defaults

    if (pooled) {
      //如果打开了这个属性
        try {
          	//获取最大字段大小
            defaultMaxFieldSize = stmt.getMaxFieldSize();
        } catch (SQLException e) {
            LOG.error("getMaxFieldSize error", e);
        }

        try {
          //获取最大行
            defaultMaxRows = stmt.getMaxRows();
        } catch (SQLException e) {
            LOG.error("getMaxRows error", e);
        }

        try {
          //获取查询超时时间
            defaultQueryTimeout = stmt.getQueryTimeout();
        } catch (SQLException e) {
            LOG.error("getMaxRows error", e);
        }

        try {
          //取数方向
            defaultFetchDirection = stmt.getFetchDirection();
        } catch (SQLException e) {
            LOG.error("getFetchDirection error", e);
        }

        try {
          //取数大小
            defaultFetchSize = stmt.getFetchSize();
        } catch (SQLException e) {
            LOG.error("getFetchSize error", e);
        }
    }

    currentMaxFieldSize = defaultMaxFieldSize;
    currentMaxRows = defaultMaxRows;
    currentQueryTimeout = defaultQueryTimeout;
    currentFetchDirection = defaultFetchDirection;
    currentFetchSize = defaultFetchSize;
}

执行查询executeQuery时序图

执行查询executeQuery源码

@Override
public ResultSet executeQuery() throws SQLException {
  //check 连接
    checkOpen();
		//执行查询的次数++
    incrementExecuteQueryCount();
   //sql 事务记录
    transactionRecord(sql);
   //oracle设置行预取
    oracleSetRowPrefetch();
	// 执行前 running状态变更
    conn.beforeExecute();
    try {
      //实际执行 PreparedStatementProxyImpl 的查询代码详解见下面
        ResultSet rs = stmt.executeQuery();

        if (rs == null) {
            return null;
        }
				//连接池返回结果封装
        DruidPooledResultSet poolableResultSet = new DruidPooledResultSet(this, rs);
      //添加结果集跟踪 用于监控
        addResultSetTrace(poolableResultSet);

        return poolableResultSet;
    } catch (Throwable t) {
        errorCheck(t);

        throw checkException(t);
    } finally {
      //更新连接的running状态
        conn.afterExecute();
    }
}

preparedStatement_executeQuery

PreparedStatementProxyImpl类对executeQuery方法的实现，这个方法实现中调用了父类StatementProxyImpl的createChain()方法，preparedStatement_executeQuery

@Override
public ResultSet executeQuery() throws SQLException {
    firstResultSet = true;

    updateCount = null;
    lastExecuteSql = sql;
    lastExecuteType = StatementExecuteType.ExecuteQuery;
    lastExecuteStartNano = -1L;
    lastExecuteTimeNano = -1L;
		// 调用父类createChain 返回FilterChainImpl对象内容，执行FilterChain的preparedStatement_executeQuery方法
    return createChain().preparedStatement_executeQuery(this);
}

FilterChainImpl

这个方法的返回值是一个过滤器链类FilterChainImpl类对象，FilterChainImpl类的

public FilterChainImpl createChain() {
  //获取FilterChainImpl对象
    FilterChainImpl chain = this.filterChain;
    if (chain == null) {
        chain = new FilterChainImpl(this.getConnectionProxy().getDirectDataSource());
    } else {
        this.filterChain = null;
    }

    return chain;
}

FilterEventAdapter类图

FilterEventAdapter源码

FilterChainImpl类的preparedStatement_executeQuery方法执行的时候会先执行nextFilter过滤器类的此方法。

@Override
public ResultSetProxy preparedStatement_executeQuery(PreparedStatementProxy statement) throws SQLException {
    if (this.pos < filterSize) {
      // 执行过滤器的方法 SQL监控的过滤器类（FilterEventAdapter）
        return nextFilter().preparedStatement_executeQuery(this, statement);
    }

    ResultSet resultSet = statement.getRawObject().executeQuery();
    if (resultSet == null) {
        return null;
    }
    return new ResultSetProxyImpl(statement, resultSet, dataSource.createResultSetId(),
            statement.getLastExecuteSql());
}

SQL监控的过滤器类（FilterEventAdapter），保存SQL执行中的监控数据。说明了druid监控数据的来源。

//FilterEventAdapter
//这个类的很巧妙的之处就是采用了设计模式中的模版方法，FilterEventAdapter作为父类实现通用的处理，子类继承这个实现具体的个性话的业务，很适合在实际业务场景中进行业务抽象模型的时候使用这种设计思路
@Override
public ResultSetProxy preparedStatement_executeQuery(FilterChain chain, PreparedStatementProxy statement)
                                                                                                         throws SQLException {
    try {
        //sql实际执行之前 调用的是 如果子类是Log Filter的时候：组装sql执行的日志  如果是Stat Filter则记录对应的监控参数
        statementExecuteQueryBefore(statement, statement.getSql());

        ResultSetProxy resultSet = chain.preparedStatement_executeQuery(statement);

        if (resultSet != null) {
            //子类中Log Filter的方法组装sql执行的日志 or Stat Filter则记录对应的监控参数
            statementExecuteQueryAfter(statement, statement.getSql(), resultSet);
            //子类中Log Filter的方法组装sql执行的日志 or Stat Filter则记录对应的监控参数
            resultSetOpenAfter(resultSet);
        }

        return resultSet;
    } catch (SQLException error) {
        statement_executeErrorAfter(statement, statement.getSql(), error);
        throw error;
    } catch (RuntimeException error) {
        statement_executeErrorAfter(statement, statement.getSql(), error);
        throw error;
    } catch (Error error) {
        statement_executeErrorAfter(statement, statement.getSql(), error);
        throw error;
    }
}

总结

今天主要针对于查询的sql在druid中具体如何执行，如果被监控，如何被记录的。通过前几天和今天的学习明白了“druid 为监控而生”的真正意义。在整个设计中监控贯穿着所有的处理，比如峰值、连接数、sql执行时间等等。在具体执行sql的时候通过Filter的方式进行拦截记录监控的相关数据。明天计划针对具体监控的StatFilter的源码进行解读。