歡迎光臨
每天分享高質量文章

資料庫中間件 MyCAT原始碼分析:【單庫單表】插入

本文主要基於 MyCAT 1.6.5 正式版

  • 1. 概述

  • 2. 接收請求,解析 SQL

  • 3. 獲得路由結果

  • 4. 獲得 MySQL 連接,執行 SQL

  • 5. 響應執行 SQL 結果


友情提示:歡迎關註公眾號【芋道原始碼】。?關註後,拉你進【原始碼圈】微信群和【芋艿】搞基嗨皮。

友情提示:歡迎關註公眾號【芋道原始碼】。?關註後,拉你進【原始碼圈】微信群和【芋艿】】搞基嗨皮。

友情提示:歡迎關註公眾號【芋道原始碼】。?關註後,拉你進【原始碼圈】微信群和【芋艿】】搞基嗨皮。


1. 概述

內容形態以 順序圖 + 核心代碼 為主。 
如果有地方表述不錯誤或者不清晰,歡迎留言。 
對於內容形態,非常糾結,如果有建議,特別特別特別歡迎您提出。 
微信號:wangwenbin-server。

本文講解 【單庫單表】插入 所涉及到的代碼。交互如下圖:

單庫單表插入簡圖

整個過程,MyCAT Server 流程如下:

  1. 接收 MySQL Client 請求,解析 SQL。

  2. 獲得路由結果,進行路由。

  3. 獲得 MySQL 連接,執行 SQL。

  4. 響應執行結果,發送結果給 MySQL Client。

我們逐個步驟分析,一起來看看原始碼。

2. 接收請求,解析 SQL

【單庫單表】插入(01主流程)

【 1 – 2 】

接收一條 MySQL 命令。在【1】之前,還有請求資料讀取、拆成單條 MySQL SQL。

【 3 】

不同 MySQL 命令,分發到不同的方法執行。核心代碼如下:

  1: // ⬇️⬇️⬇️【FrontendCommandHandler.java】
 2: public class FrontendCommandHandler implements NIOHandler {
 3:
 4:     @Override
 5:     public void handle(byte[] data) {
 6:    
 7:         // .... 省略部分代碼
 8:         switch (data[4]) //
 9:         {
10:             case MySQLPacket.COM_INIT_DB:
11:                 commands.doInitDB();
12:                 source.initDB(data);
13:                 break;
14:             case MySQLPacket.COM_QUERY: // 查詢命令
15:                 // 計數查詢命令
16:                 commands.doQuery();
17:                 // 執行查詢命令
18:                 source.query(data);
19:                 break;
20:             case MySQLPacket.COM_PING:
21:                 commands.doPing();
22:                 source.ping();
23:                 break;
24:             // .... 省略部分case
25:         }
26:     }
27:
28: }

INSERT/SELECT/UPDATE/DELETE 等 SQL 歸屬於 MySQLPacket.COM_QUERY,詳細可見:《MySQL協議分析#4.2 客戶端命令請求報文(客戶端 -> 服務器)》。

【 4 】

將 二進制陣列 解析成 SQL。核心代碼如下:

  1: // ⬇️⬇️⬇️【FrontendConnection.java】
 2: public void query(byte[] data) {
 3:     // 取得陳述句
 4:     String sql = null;      
 5:     try {
 6:         MySQLMessage mm = new MySQLMessage(data);
 7:         mm.position(5);
 8:         sql = mm.readString(charset);
 9:     } catch (UnsupportedEncodingException e) {
10:         writeErrMessage(ErrorCode.ER_UNKNOWN_CHARACTER_SET, "Unknown charset '" + charset + "'");
11:         return;
12:     }      
13:     // 執行陳述句
14:     this.query( sql );
15: }

【 5 】

解析 SQL 型別。核心代碼如下:

  1: // ⬇️⬇️⬇️【ServerQueryHandler.java】
 2: @Override
 3: public void query(String sql) {
 4:     // 解析 SQL 型別
 5:     int rs = ServerParse.parse(sql);
 6:     int sqlType = rs & 0xff;
 7:    
 8:     switch (sqlType) {
 9:     //explain sql
10:     case ServerParse.EXPLAIN:
11:         ExplainHandler.handle(sql, c, rs >>> 8);
12:         break;
13:     // .... 省略部分case
14:         break;
15:     case ServerParse.SELECT:
16:         SelectHandler.handle(sql, c, rs >>> 8);
17:         break;
18:     // .... 省略部分case
19:     default:
20:         if(readOnly){
21:             LOGGER.warn(new StringBuilder().append("User readonly:").append(sql).toString());
22:             c.writeErrMessage(ErrorCode.ER_USER_READ_ONLY, "User readonly");
23:             break;
24:         }
25:         c.execute(sql, rs & 0xff);
26:     }
27: }
28:
29:
30: // ⬇️⬇️⬇️【ServerParse.java】
31: public static int parse(String stmt) {
32:     int length = stmt.length();
33:     //FIX BUG FOR SQL SUCH AS /XXXX/SQL
34:     int rt = -1;
35:     for (int i = 0; i < length; ++i) {
36:         switch (stmt.charAt(i)) {
37:         // .... 省略部分case            case 'I':
38:         case 'i':
39:             rt = insertCheck(stmt, i);
40:             if (rt != OTHER) {
41:                 return rt;
42:             }
43:             continue;
44:             // .... 省略部分case
45:         case 'S':
46:         case 's':
47:             rt = sCheck(stmt, i);
48:             if (rt != OTHER) {
49:                 return rt;
50:             }
51:             continue;
52:             // .... 省略部分case
53:         default:
54:             continue;
55:         }
56:     }
57:     return OTHER;
58: }

【 6 】

執行 SQL,詳細解析見下文,核心代碼如下:

  1: // ⬇️⬇️⬇️【ServerConnection.java】
 2: public class ServerConnection extends FrontendConnection {
 3:     public void execute(String sql, int type) {
 4:         // .... 省略代碼
 5:         SchemaConfig schema = MycatServer.getInstance().getConfig().getSchemas().get(db);
 6:         if (schema == null) {
 7:             writeErrMessage(ErrorCode.ERR_BAD_LOGICDB,
 8:                     "Unknown MyCAT Database '" + db + "'");
 9:             return;
10:         }
11:
12:         // .... 省略代碼
13:
14:         // 路由到後端資料庫,執行 SQL
15:         routeEndExecuteSQL(sql, type, schema);
16:     }
17:    
18:     public void routeEndExecuteSQL(String sql, final int type, final SchemaConfig schema) {
19:         // 路由計算
20:         RouteResultset rrs = null;
21:         try {
22:             rrs = MycatServer
23:                     .getInstance()
24:                     .getRouterservice()
25:                     .route(MycatServer.getInstance().getConfig().getSystem(),
26:                             schema, type, sql, this.charset, this);
27:
28:         } catch (Exception e) {
29:             StringBuilder s = new StringBuilder();
30:             LOGGER.warn(s.append(this).append(sql).toString() + " err:" + e.toString(),e);
31:             String msg = e.getMessage();
32:             writeErrMessage(ErrorCode.ER_PARSE_ERROR, msg == null ? e.getClass().getSimpleName() : msg);
33:             return;
34:         }
35:
36:         // 執行 SQL
37:         if (rrs != null) {
38:             // session執行
39:             session.execute(rrs, rrs.isSelectForUpdate() ? ServerParse.UPDATE : type);
40:         }
41:        
42:      }
43:
44: }

3. 獲得路由結果

【單庫單表】插入(02獲取路由)

【 1 – 2 】【 12 】

獲得路由主流程。核心代碼如下:

  1: // ⬇️⬇️⬇️【RouteService.java】
 2: public RouteResultset route(SystemConfig sysconf, SchemaConfig schema,
 3:         int sqlType, String stmt, String charset, ServerConnection sc)

 4:         throws SQLNonTransientException
{
 5:     RouteResultset rrs = null;
 6:     // .... 省略代碼
 7:     int hintLength = RouteService.isHintSql(stmt);
 8:     if(hintLength != -1){ // TODO 待讀:hint
 9:         // .... 省略代碼
10:         }
11:     } else {
12:         stmt = stmt.trim();
13:         rrs = RouteStrategyFactory.getRouteStrategy().route(sysconf, schema, sqlType, stmt,
14:                 charset, sc, tableId2DataNodeCache);
15:     }
16:
17:     // .... 省略代碼        return rrs;
18: }
19: // ⬇️⬇️⬇️【AbstractRouteStrategy.java】
20: @Override
21: public RouteResultset route(SystemConfig sysConfig, SchemaConfig schema, int sqlType, String origSQL,
22:         String charset, ServerConnection sc, LayerCachePool cachePool)
throws SQLNonTransientException
{
23:
24:     // .... 省略代碼
25:
26:     // 處理一些路由之前的邏輯;全域性序列號,父子表插入
27:     if (beforeRouteProcess(schema, sqlType, origSQL, sc) ) {
28:         return null;
29:     }
30:
31:     // .... 省略代碼
32:
33:     // 檢查是否有分片
34:     if (schema.isNoSharding() && ServerParse.SHOW != sqlType) {
35:         rrs = RouterUtil.routeToSingleNode(rrs, schema.getDataNode(), stmt);
36:     } else {
37:         RouteResultset returnedSet = routeSystemInfo(schema, sqlType, stmt, rrs);
38:         if (returnedSet == null) {
39:             rrs = routeNormalSqlWithAST(schema, stmt, rrs, charset, cachePool,sqlType,sc);
40:         }
41:     }
42:
43:     return rrs;
44: }

路由 詳細解析,我們另開文章,避免內容過多,影響大家對【插入】流程和邏輯的理解。

【 3 – 6 】

路由前置處理。當符合如下三種情況下,進行處理:

{ 1 } 使用全域性序列號

insert into table (id, name) values (NEXT VALUE FOR MYCATSEQ_ID, 'name')

{ 2 } ER 子表插入 
{ 3 } 主鍵使用自增 ID 插入:

insert into table (name) values ('name')
===>
insert into table (id, name) values (NEXT VALUE FOR MYCATSEQ_ID, 'name')

情況 { 1 } { 3 } 情況類似,使用全域性序列號。

核心代碼如下:

  1: // ⬇️⬇️⬇️【AbstractRouteStrategy.java】
 2: private boolean beforeRouteProcess(SchemaConfig schema, int sqlType, String origSQL, ServerConnection sc)
 3:         throws SQLNonTransientException
{
 4:     return  // 處理 id 使用 全域性序列號
 5:             RouterUtil.processWithMycatSeq(schema, sqlType, origSQL, sc)
 6:             // 處理 ER 子表
 7:             || (sqlType == ServerParse.INSERT && RouterUtil.processERChildTable(schema, origSQL, sc))
 8:             // 處理 id 自增長
 9:             || (sqlType == ServerParse.INSERT && RouterUtil.processInsert(schema, sqlType, origSQL, sc));
10: }

RouterUtil.java 處理 SQL 考慮性能,實現會比較 C-style,代碼咱就不貼了,傳送門:https://github.com/YunaiV/Mycat-Server/blob/1.6/src/main/java/io/mycat/route/util/RouterUtil.java。 (?該倉庫從官方 Fork,逐步完善中文註釋,歡迎 Star)

【 7 – 11 】

前置路由處理全域性序列號時,添加到全域性序列處理器(MyCATSequnceProcessor)。該處理器會異步生成 ID,替換 SQL 內的 NEXT VALUE FOR MYCATSEQ_ 正則。例如:

insert into table (id, name) values (NEXT VALUE FOR MYCATSEQ_ID, 'name')
===>
insert into table (id, name) values (868348974560579584, 'name')

異步處理完後,呼叫 ServerConnection#routeEndExecuteSQL(sql, type, schema) 方法重新執行 SQL。

核心代碼如下:

  1: // ⬇️⬇️⬇️【RouterUtil.java】
 2: public static void processSQL(ServerConnection sc,SchemaConfig schema,String sql,int sqlType){
 3:     SessionSQLPair sessionSQLPair = new SessionSQLPair(sc.getSession2(), schema, sql, sqlType);
 4:     MycatServer.getInstance().getSequnceProcessor().addNewSql(sessionSQLPair);
 5: }
 6: // ⬇️⬇️⬇️【MyCATSequnceProcessor.java】
 7: public class MyCATSequnceProcessor {
 8:     private LinkedBlockingQueue seqSQLQueue = new LinkedBlockingQueue();
 9:     private volatile boolean running=true;
10:    
11:     public void addNewSql(SessionSQLPair pair) {
12:         seqSQLQueue.add(pair);
13:     }
14:
15:     private void executeSeq(SessionSQLPair pair) {
16:         try {
17:            
18:             // 使用Druid解析器實現sequence處理  @兵臨城下
19:             DruidSequenceHandler sequenceHandler = new DruidSequenceHandler(MycatServer
20:                     .getInstance().getConfig().getSystem().getSequnceHandlerType());
21:
22:             // 生成可執行 SQL :目前主要是生成 id
23:             String charset = pair.session.getSource().getCharset();
24:             String executeSql = sequenceHandler.getExecuteSql(pair.sql,charset == null ? "utf-8":charset);
25:
26:             // 執行 SQL
27:             pair.session.getSource().routeEndExecuteSQL(executeSql, pair.type,pair.schema);
28:         } catch (Exception e) {
29:             LOGGER.error("MyCATSequenceProcessor.executeSeq(SesionSQLPair)",e);
30:             pair.session.getSource().writeErrMessage(ErrorCode.ER_YES,"mycat sequnce err." + e);
31:             return;
32:         }
33:     }
34:    
35:     class ExecuteThread extends Thread {
36:        
37:         public ExecuteThread() {
38:             setDaemon(true); // 設置為後臺執行緒,防止throw RuntimeExecption行程仍然存在的問題
39:         }
40:        
41:         public void run() {
42:             while (running) {
43:                 try {
44:                     SessionSQLPair pair=seqSQLQueue.poll(100,TimeUnit.MILLISECONDS);
45:                     if(pair!=null){
46:                         executeSeq(pair);
47:                     }
48:                 } catch (Exception e) {
49:                     LOGGER.warn("MyCATSequenceProcessor$ExecutorThread",e);
50:                 }
51:             }
52:         }
53:     }
54: }

❓此處有個疑問:MyCATSequnceProcessor 是單執行緒,會不會插入性能有一定的影響?後續咱做下性能測試。

4. 獲得 MySQL 連接,執行 SQL

【單庫單表】插入(03執行 SQL)

【 1 – 8 】

獲得 MySQL 連接。

  • PhysicalDBNode :物理資料庫節點。

  • PhysicalDatasource :物理資料庫資料源。

【 9 – 13 】

發送 SQL 到 MySQL Server,執行 SQL。

5. 響應執行 SQL 結果

【單庫單表】插入(04執行響應)

【 1 – 4 】

處理 MySQL Server 響應資料包。

【 5 – 8 】

發送插入成功結果給 MySQL Client。

赞(0)

分享創造快樂