MyCAT PreparedStatement 從新入門

1. 概述

相信不少同窗在學習 JDBC 時，都碰到 PreparedStatement 和 Statement。究竟該使用哪一個呢？最終極可能是懵裏懵懂的看了各類總結，使用 PreparedStatement。那麼本文，經過 MyCAT 對PreparedStatement 的實現對你們可以從新理解下。

本文主要分紅兩部分：

1. JDBC Client 如何實現 PreparedStatement。
2. MyCAT Server 如何處理 PreparedStatement。

2. JDBC Client 實現

首先，咱們來看一段你們最喜歡複製粘貼之一的代碼，JDBC PreparedStatement 查詢 MySQL 數據庫：

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 解析開銷。

【前者】性能必定比【後者】好嗎？相信你已經有了正確的答案。

3. MyCAT Server 實現

3.1 建立 PreparedStatement

該操做對應 Client conn.prepareStatement(....)。

MyCAT 接收到請求後，建立 PreparedStatement，並返回 statementId 等信息。Client 發起 SQL 執行時，須要將 statementId 帶給 MyCAT。核心代碼以下：

// 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();
}