title: MyCAT 源码分析 —— 【单库单表】查询
date: 2017-05-30
tags:
categories: MyCAT
permalink: MyCAT/single-db-single-table-select

&＃xfffd;&＃xfffd;&＃xfffd;&＃xfffd;&＃xfffd;&＃xfffd;关注微信公众号&＃xff1a;【芋艿的后端小屋】有福利&＃xff1a;
1. RocketMQ / MyCAT / Sharding-JDBC 所有源码分析文章列表
2. RocketMQ / MyCAT / Sharding-JDBC 中文注释源码 GitHub 地址
3. 您对于源码的疑问每条留言都将得到认真回复。甚至不知道如何读源码也可以请教噢。
4. 新的源码解析文章实时收到通知。每周更新一篇左右。

• 1. 概述
• 2. 接收请求&＃xff0c;解析 SQL
• 3. 获得路由结果
• 4. 获得 MySQL 连接&＃xff0c;执行 SQL
• 5. 响应执行 SQL 结果
• 6. 其他 &＃xff1a;更新 / 删除

1. 概述

内容形态以 顺序图 &＃43; 核心代码 为主。
如果有地方表述不错误或者不清晰&＃xff0c;欢迎留言。
对于内容形态&＃xff0c;非常纠结&＃xff0c;如果有建议&＃xff0c;特别特别特别欢迎您提出。
微信号&＃xff1a;wangwenbin-server。

本文讲解 【单库单表】查询 所涉及到的代码。

&＃xfffd;&＃xfffd;内容和 《MyCAT 源码分析 —— 【单库单表】插入》 超级相似&＃xff0c;一方面本身流程基本相同&＃xff0c;另外一方面文章结构没拆分好。我们使用 &＃xfffd;&＃xfffd; 标记差异的逻辑。

交互如下图&＃xff1a;

整个过程&＃xff0c;MyCAT Server 流程如下&＃xff1a;

1. 接收 MySQL Client 请求&＃xff0c;解析 SQL。
2. 获得路由结果&＃xff0c;进行路由。
3. 获得 MySQL 连接&＃xff0c;执行 SQL。
4. 响应执行结果&＃xff0c;发送结果给 MySQL Client。

我们逐个步骤分析&＃xff0c;一起来看看源码。

2. 接收请求&＃xff0c;解析 SQL

【1 - 2】

接收一条 MySQL 命令。在【1】之前&＃xff0c;还有请求数据读取、拆成单条 MySQL SQL。

【3】

1: // ⬇️⬇️⬇️【FrontendCommandHandler.java】2: public class FrontendCommandHandler implements NIOHandler {3: 4: &＃64;Override5: 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: // .... 省略部分case25: }26: }27: 28: }

INSERT/SELECT/UPDATE/DELETE 等 SQL 归属于 MySQLPacket.COM_QUERY&＃xff0c;详细可见&＃xff1a;《MySQL协议分析#4.2 客户端命令请求报文&＃xff08;客户端 -> 服务器&＃xff09;》。

【4】

将 二进制数组 解析成 SQL。核心代码如下&＃xff1a;

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

【5】

解析 SQL 类型。核心代码如下&＃xff1a;

1: // ⬇️⬇️⬇️【ServerQueryHandler.java】2: &＃64;Override3: public void query(String sql) {4: // 解析 SQL 类型5: int rs &＃61; ServerParse.parse(sql);6: int sqlType &＃61; rs & 0xff;7: 8: switch (sqlType) {9: //explain sql10: case ServerParse.EXPLAIN:11: ExplainHandler.handle(sql, c, rs >>> 8);12: break;13: // .... 省略部分case14: break;15: case ServerParse.SELECT:16: SelectHandler.handle(sql, c, rs >>> 8);17: break;18: // .... 省略部分case19: 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 &＃61; stmt.length();33: //FIX BUG FOR SQL SUCH AS /XXXX/SQL34: int rt &＃61; -1;35: for (int i &＃61; 0; i 36: switch (stmt.charAt(i)) {37: // .... 省略部分case case &＃39;I&＃39;:38: case &＃39;i&＃39;:39: rt &＃61; insertCheck(stmt, i);40: if (rt !&＃61; OTHER) {41: return rt;42: }43: continue;44: // .... 省略部分case45: case &＃39;S&＃39;:46: case &＃39;s&＃39;:47: rt &＃61; sCheck(stmt, i);48: if (rt !&＃61; OTHER) {49: return rt;50: }51: continue;52: // .... 省略部分case53: default:54: continue;55: }56: }57: return OTHER;58: }

&＃xfffd;&＃xfffd;【6】【7】

解析 Select SQL 类型&＃xff0c;分发到对应的逻辑。核心代码如下&＃xff1a;

1: // ⬇️⬇️⬇️【SelectHandler.java】2: public static void handle(String stmt, ServerConnection c, int offs) {3: int offset &＃61; offs;4: switch (ServerParseSelect.parse(stmt, offs)) { // 解析 Select SQL 类型5: case ServerParseSelect.VERSION_COMMENT: // select &＃64;&＃64;VERSION_COMMENT;6: SelectVersionComment.response(c);7: break;8: case ServerParseSelect.DATABASE: // select DATABASE();9: SelectDatabase.response(c);10: break;11: case ServerParseSelect.USER: // select CURRENT_USER();12: SelectUser.response(c);13: break;14: case ServerParseSelect.VERSION: // select VERSION();15: SelectVersion.response(c);16: break;17: case ServerParseSelect.SESSION_INCREMENT: // select &＃64;&＃64;session.auto_increment_increment;18: SessionIncrement.response(c);19: break;20: case ServerParseSelect.SESSION_ISOLATION: // select &＃64;&＃64;session.tx_isolation;21: SessionIsolation.response(c);22: break;23: case ServerParseSelect.LAST_INSERT_ID: // select LAST_INSERT_ID();24: // ....省略代码25: break;26: case ServerParseSelect.IDENTITY: // select &＃64;&＃64;identity27: // ....省略代码28: break;29: case ServerParseSelect.SELECT_VAR_ALL: //30: SelectVariables.execute(c,stmt);31: break;32: case ServerParseSelect.SESSION_TX_READ_ONLY: //33: SelectTxReadOnly.response(c);34: break;35: default: // 其他&＃xff0c;例如 select * from table36: c.execute(stmt, ServerParse.SELECT);37: }38: }39: // ⬇️⬇️⬇️【ServerParseSelect.java】40: public static int parse(String stmt, int offset) {41: int i &＃61; offset;42: for (; i 43: switch (stmt.charAt(i)) {44: case &＃39; &＃39;:45: continue;46: case &＃39;/&＃39;:47: case &＃39;#&＃39;:48: i &＃61; ParseUtil.comment(stmt, i);49: continue;50: case &＃39;&＃64;&＃39;:51: return select2Check(stmt, i);52: case &＃39;D&＃39;:53: case &＃39;d&＃39;:54: return databaseCheck(stmt, i);55: case &＃39;L&＃39;:56: case &＃39;l&＃39;:57: return lastInsertCheck(stmt, i);58: case &＃39;U&＃39;:59: case &＃39;u&＃39;:60: return userCheck(stmt, i);61: case &＃39;C&＃39;:62: case &＃39;c&＃39;:63: return currentUserCheck(stmt, i);64: case &＃39;V&＃39;:65: case &＃39;v&＃39;:66: return versionCheck(stmt, i);67: default:68: return OTHER;69: }70: }71: return OTHER;72: }

【8】

执行 SQL&＃xff0c;详细解析见下文&＃xff0c;核心代码如下&＃xff1a;

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

3. 获得路由结果

【 1 - 5 】

获得路由主流程。核心代码如下&＃xff1a;

1: // ⬇️⬇️⬇️【SelectHandler.java】2: public RouteResultset route(SystemConfig sysconf, SchemaConfig schema,3: int sqlType, String stmt, String charset, ServerConnection sc)4: throws SQLNonTransientException {5: RouteResultset rrs &＃61; null;6: 7: // SELECT 类型的SQL, 检测缓存是否存在8: if (sqlType &＃61;&＃61; ServerParse.SELECT) {9: cacheKey &＃61; schema.getName() &＃43; stmt; 10: rrs &＃61; (RouteResultset) sqlRouteCache.get(cacheKey);11: if (rrs !&＃61; null) {12: checkMigrateRule(schema.getName(),rrs,sqlType);13: return rrs;14: }15: }16: }17: 18: // .... 省略代码19: int hintLength &＃61; RouteService.isHintSql(stmt);20: if(hintLength !&＃61; -1){ // TODO 待读&＃xff1a;hint21: // .... 省略代码22: }23: } else {24: stmt &＃61; stmt.trim();25: rrs &＃61; RouteStrategyFactory.getRouteStrategy().route(sysconf, schema, sqlType, stmt,26: charset, sc, tableId2DataNodeCache);27: }28: 29: // 记录查询命令路由结果缓存30: if (rrs !&＃61; null && sqlType &＃61;&＃61; ServerParse.SELECT && rrs.isCacheAble()) {31: sqlRouteCache.putIfAbsent(cacheKey, rrs);32: }33: // .... 省略代码 return rrs;34: }35: // ⬇️⬇️⬇️【AbstractRouteStrategy.java】36: &＃64;Override37: public RouteResultset route(SystemConfig sysConfig, SchemaConfig schema, int sqlType, String origSQL,38: String charset, ServerConnection sc, LayerCachePool cachePool) throws SQLNonTransientException {39: 40: // .... 省略代码41: 42: // 处理一些路由之前的逻辑;全局序列号&＃xff0c;父子表插入43: if (beforeRouteProcess(schema, sqlType, origSQL, sc) ) {44: return null;45: }46: 47: // .... 省略代码48: 49: // 检查是否有分片50: if (schema.isNoSharding() && ServerParse.SHOW !&＃61; sqlType) {51: rrs &＃61; RouterUtil.routeToSingleNode(rrs, schema.getDataNode(), stmt);52: } else {53: RouteResultset returnedSet &＃61; routeSystemInfo(schema, sqlType, stmt, rrs);54: if (returnedSet &＃61;&＃61; null) {55: rrs &＃61; routeNormalSqlWithAST(schema, stmt, rrs, charset, cachePool,sqlType,sc);56: }57: }58: 59: return rrs;60: }

&＃xfffd;&＃xfffd;【3】第 7 至 16 行 &＃xff1a;当 Select SQL 存在路由结果缓存时&＃xff0c;直接返回缓存。
&＃xfffd;&＃xfffd;【6】第 29 至 32 行 &＃xff1a;记录 Select SQL 路由结果到缓存。

路由 详细解析&＃xff0c;我们另开文章&＃xff0c;避免内容过多&＃xff0c;影响大家对【插入】流程和逻辑的理解。

4. 获得 MySQL 连接&＃xff0c;执行 SQL

【 1 - 8 】

获得 MySQL 连接。

• PhysicalDBNode &＃xff1a;物理数据库节点。
• PhysicalDatasource &＃xff1a;物理数据库数据源。

【 9 - 13 】

发送 SQL 到 MySQL Server&＃xff0c;执行 SQL。

&＃xfffd;&＃xfffd; 5. 响应执行 SQL 结果

核心代码如下&＃xff1a;

1: // ⬇️⬇️⬇️【MySQLConnectionHandler.java】2: &＃64;Override3: protected void handleData(byte[] data) {4: switch (resultStatus) {5: case RESULT_STATUS_INIT:6: switch (data[4]) {7: case OkPacket.FIELD_COUNT:8: handleOkPacket(data);9: break;10: case ErrorPacket.FIELD_COUNT:11: handleErrorPacket(data);12: break;13: case RequestFilePacket.FIELD_COUNT:14: handleRequestPacket(data);15: break;16: default: // 初始化 header fields17: resultStatus &＃61; RESULT_STATUS_HEADER;18: header &＃61; data;19: fields &＃61; new ArrayList((int) ByteUtil.readLength(data,20: 4));21: }22: break;23: case RESULT_STATUS_HEADER:24: switch (data[4]) {25: case ErrorPacket.FIELD_COUNT:26: resultStatus &＃61; RESULT_STATUS_INIT;27: handleErrorPacket(data);28: break;29: case EOFPacket.FIELD_COUNT: // 解析 fields 结束30: resultStatus &＃61; RESULT_STATUS_FIELD_EOF;31: handleFieldEofPacket(data);32: break;33: default: // 解析 fields34: fields.add(data);35: }36: break;37: case RESULT_STATUS_FIELD_EOF:38: switch (data[4]) {39: case ErrorPacket.FIELD_COUNT:40: resultStatus &＃61; RESULT_STATUS_INIT;41: handleErrorPacket(data);42: break;43: case EOFPacket.FIELD_COUNT: // 解析 每行记录 结束44: resultStatus &＃61; RESULT_STATUS_INIT;45: handleRowEofPacket(data);46: break;47: default: // 每行记录48: handleRowPacket(data);49: }50: break;51: default:52: throw new RuntimeException("unknown status!");53: }54: }

6. 其他 &＃xff1a;更新 / 删除

流程基本和 《MyCAT源码分析&＃xff1a;【单库单表】插入》 相同。我们就不另外文章解析。