MyCAT PreparedStatement 从新入门

时间 2019-11-13

标签 mycat preparedstatement 从新入门繁體版

原文原文链接

1. 概述

相信不少同窗在学习 JDBC 时，都碰到 PreparedStatement 和 Statement。究竟该使用哪一个呢？最终极可能是懵里懵懂的看了各类总结，使用 PreparedStatement。那么本文，经过 MyCAT 对PreparedStatement 的实现对你们可以从新理解下。java

本文主要分红两部分：mysql

JDBC Client 如何实现 PreparedStatement。
MyCAT Server 如何处理 PreparedStatement。

2. JDBC Client 实现

首先，咱们来看一段你们最喜欢复制粘贴之一的代码，JDBC PreparedStatement 查询 MySQL 数据库：sql

public class PreparedStatementDemo {
    public static void main(String[] args) throws ClassNotFoundException, SQLException {
        // 1. 得到数据库链接
        Class.forName("com.mysql.jdbc.Driver");
        Connection conn = DriverManager.getConnection("jdbc:mysql://127.0.0.1:8066/dbtest?useServerPrepStmts=true", "root", "123456");
        // PreparedStatement
        PreparedStatement ps = conn.prepareStatement("SELECT id, username, password FROM t_user WHERE id = ?");
        ps.setLong(1, Math.abs(new Random().nextLong()));
        // execute
        ps.executeQuery();
    }
}

获取 MySQL 链接时，useServerPrepStmts=true 是很是很是很是重要的参数。若是不配置，PreparedStatement 实际是个假的 PreparedStatement（新版本默认为 FALSE，听说部分老版本默认为 TRUE），未开启服务端级别的 SQL 预编译。数据库

WHY ？来看下 JDBC 里面是怎么实现的。缓存

// com.mysql.jdbc.ConnectionImpl.java
public PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
   synchronized (getConnectionMutex()) {
       checkClosed();
       
       PreparedStatement pStmt = null;
       boolean canServerPrepare = true;
       String nativeSql = getProcessEscapeCodesForPrepStmts() ? nativeSQL(sql) : sql;
       if (this.useServerPreparedStmts && getEmulateUnsupportedPstmts()) {
           canServerPrepare = canHandleAsServerPreparedStatement(nativeSql);
       }
       if (this.useServerPreparedStmts && canServerPrepare) {
           if (this.getCachePreparedStatements()) { // 从缓存中获取 pStmt
               synchronized (this.serverSideStatementCache) {
                   pStmt = (com.mysql.jdbc.ServerPreparedStatement) this.serverSideStatementCache
                           .remove(makePreparedStatementCacheKey(this.database, sql));
                   if (pStmt != null) {
                       ((com.mysql.jdbc.ServerPreparedStatement) pStmt).setClosed(false);
                       pStmt.clearParameters(); // 清理上次留下的参数
                   }
                   if (pStmt == null) {
                        // .... 省略代码 ：向 Server 提交 SQL 预编译。
                   }
               }
           } else {
               try {
                   // 向 Server 提交 SQL 预编译。
                   pStmt = ServerPreparedStatement.getInstance(getMultiHostSafeProxy(), nativeSql, this.database, resultSetType, resultSetConcurrency);
                   pStmt.setResultSetType(resultSetType);
                   pStmt.setResultSetConcurrency(resultSetConcurrency);
               } catch (SQLException sqlEx) {
                   // Punt, if necessary
                   if (getEmulateUnsupportedPstmts()) {
                       pStmt = (PreparedStatement) clientPrepareStatement(nativeSql, resultSetType, resultSetConcurrency, false);
                   } else {
                       throw sqlEx;
                   }
               }
           }
       } else {
           pStmt = (PreparedStatement) clientPrepareStatement(nativeSql, resultSetType, resultSetConcurrency, false);
       }
       return pStmt;
   }
}

【前者】当 Client 开启 useServerPreparedStmts 而且 Server 支持 ServerPrepare，Client 会向 Server 提交 SQL 预编译请求。

if (this.useServerPreparedStmts && canServerPrepare) {
    pStmt = ServerPreparedStatement.getInstance(getMultiHostSafeProxy(), nativeSql, this.database, resultSetType, resultSetConcurrency);
}

【后者】当 Client 未开启 useServerPreparedStmts 或者 Server 不支持 ServerPrepare，Client 建立 PreparedStatement，不会向 Server 提交 SQL 预编译请求。

pStmt = (PreparedStatement) clientPrepareStatement(nativeSql, resultSetType, resultSetConcurrency, false);

即便这样，究竟为何性能会更好呢？网络

【前者】返回的 PreparedStatement 对象类是 JDBC42ServerPreparedStatement.java，后续每次执行 SQL 只需将对应占位符?对应的值提交给 Server便可，减小网络传输和 SQL 解析开销。
【后者】返回的 PreparedStatement 对象类是 JDBC42PreparedStatement.java，后续每次执行 SQL 须要将完整的 SQL 提交给 Server，增长了网络传输和 SQL 解析开销。

【前者】性能必定比【后者】好吗？相信你已经有了正确的答案。app

3. MyCAT Server 实现

3.1 建立 PreparedStatement

该操做对应 Client conn.prepareStatement(....)。dom

MyCAT 接收到请求后，建立 PreparedStatement，并返回 statementId 等信息。Client 发起 SQL 执行时，须要将 statementId 带给 MyCAT。核心代码以下：ide

// ServerPrepareHandler.java
@Override
public void prepare(String sql) {
LOGGER.debug("use server prepare, sql: " + sql);
   PreparedStatement pstmt = pstmtForSql.get(sql);
   if (pstmt == null) { // 缓存中获取
   	// 解析获取字段个数和参数个数
   	int columnCount = getColumnCount(sql);
   	int paramCount = getParamCount(sql);
       pstmt = new PreparedStatement(++pstmtId, sql, columnCount, paramCount);
       pstmtForSql.put(pstmt.getStatement(), pstmt);
       pstmtForId.put(pstmt.getId(), pstmt);
   }
   PreparedStmtResponse.response(pstmt, source);
}
// PreparedStmtResponse.java
public static void response(PreparedStatement pstmt, FrontendConnection c) {
   byte packetId = 0;
   // write preparedOk packet
   PreparedOkPacket preparedOk = new PreparedOkPacket();
   preparedOk.packetId = ++packetId;
   preparedOk.statementId = pstmt.getId();
   preparedOk.columnsNumber = pstmt.getColumnsNumber();
   preparedOk.parametersNumber = pstmt.getParametersNumber();
   ByteBuffer buffer = preparedOk.write(c.allocate(), c,true);
   // write parameter field packet
   int parametersNumber = preparedOk.parametersNumber;
   if (parametersNumber > 0) {
       for (int i = 0; i < parametersNumber; i++) {
           FieldPacket field = new FieldPacket();
           field.packetId = ++packetId;
           buffer = field.write(buffer, c,true);
       }
       EOFPacket eof = new EOFPacket();
       eof.packetId = ++packetId;
       buffer = eof.write(buffer, c,true);
   }
   // write column field packet
   int columnsNumber = preparedOk.columnsNumber;
   if (columnsNumber > 0) {
       for (int i = 0; i < columnsNumber; i++) {
           FieldPacket field = new FieldPacket();
           field.packetId = ++packetId;
           buffer = field.write(buffer, c,true);
       }
       EOFPacket eof = new EOFPacket();
       eof.packetId = ++packetId;
       buffer = eof.write(buffer, c,true);
   }
   // send buffer
   c.write(buffer);
}

每一个链接之间，PreparedStatement 不共享，即不一样链接，即便 SQL相同，对应的 PreparedStatement 不一样。性能

3.2 执行 SQL

该操做对应 Client conn.execute(....)。

MyCAT 接收到请求后，将 PreparedStatement 使用请求的参数格式化成可执行的 SQL 进行执行。伪代码以下：

String sql = pstmt.sql.format(request.params);
execute(sql);

核心代码以下：

// ServerPrepareHandler.java
@Override
public void execute(byte[] data) {
   long pstmtId = ByteUtil.readUB4(data, 5);
   PreparedStatement pstmt = null;
   if ((pstmt = pstmtForId.get(pstmtId)) == null) {
       source.writeErrMessage(ErrorCode.ER_ERROR_WHEN_EXECUTING_COMMAND, "Unknown pstmtId when executing.");
   } else {
       // 参数读取
       ExecutePacket packet = new ExecutePacket(pstmt);
       try {
           packet.read(data, source.getCharset());
       } catch (UnsupportedEncodingException e) {
           source.writeErrMessage(ErrorCode.ER_ERROR_WHEN_EXECUTING_COMMAND, e.getMessage());
           return;
       }
       BindValue[] bindValues = packet.values;
       // 还原sql中的动态参数为实际参数值
       String sql = prepareStmtBindValue(pstmt, bindValues);
       // 执行sql
       source.getSession2().setPrepared(true);
       source.query(sql);
   }
}
private String prepareStmtBindValue(PreparedStatement pstmt, BindValue[] bindValues) {
   String sql = pstmt.getStatement();
   int[] paramTypes = pstmt.getParametersType();
   StringBuilder sb = new StringBuilder();
   int idx = 0;
   for (int i = 0, len = sql.length(); i < len; i++) {
       char c = sql.charAt(i);
       if (c != '?') {
           sb.append(c);
           continue;
       }
       // 处理占位符?
       int paramType = paramTypes[idx];
       BindValue bindValue = bindValues[idx];
       idx++;
       // 处理字段为空的状况
       if (bindValue.isNull) {
           sb.append("NULL");
           continue;
       }
       // 非空状况, 根据字段类型获取值
       switch (paramType & 0xff) {
           case Fields.FIELD_TYPE_TINY:
               sb.append(String.valueOf(bindValue.byteBinding));
               break;
           case Fields.FIELD_TYPE_SHORT:
               sb.append(String.valueOf(bindValue.shortBinding));
               break;
           case Fields.FIELD_TYPE_LONG:
               sb.append(String.valueOf(bindValue.intBinding));
               break;
           // .... 省略非核心代码
        }
   }
   return sb.toString();
}