2024-01-18  阅读(62)
原文作者:hashcon 原文地址: https://zhanghaoxin.blog.csdn.net/article/details/51365041

5. 路由模块

5.1 路由模块组成:

路由模块,我们可以先把他当做个黑盒,看下输入和输出都是神马。

202401182020380001.png
输入,很明显,就是个SQL语句,原生的,不加任何修饰的,纯洁的,从客户端发过来刚刚被解码的SQL语句。
输出呢?就是个优化,改写后的SQL语句,以及要发送到的后台分片。
这个RouteResultSet就是输出,长什么样子呢?
下图是主要涉及到的类:

202401182020383462.png

  • RouteResultSet:

    • sqlType:SQL类型(select?insert?…)
    • nodes: 语句和Datanode对应关系。一条语句可以根据不同节点拆成多条不同语句
    • subTables:分表,1.6后功能,单node多表
    • sqlStatement:经过DruidParser解析后的语句
    • limitStart,limitSize:含有limit的SQL的起始点和长度
    • cacheAble:是否可以缓存(MyCat缓存中会保存SQL(key)->RouteResultSet(value))
    • primaryKey:为了实现以后完整的主键缓存而预留
    • sqlMerge:带有合并函数的sql语句处理类
    • callStatement:是否为调用存储过程的语句(call)
    • globalTableFlag:操作表是否包含全局表
    • isFinishedRoute:是否路由完成
    • autocommit:是否为自动提交
    • isLoadData:是否是LoadData命令
    • canRunInReadDb:是否能在读节点上运行
    • runOnSlave:是否在从节点上运行
    • procedure:调用存储过程处理类
  • RouteResultSetNode:

    • serialVersionUID:全局序列化类版本标识
    • name:数据节点名称
    • statement:实际执行的语句
    • srcStatement:源语句
    • sqlType:sql类型
    • canRunInReadDB:是否可以在读节点运行
    • hasBlanceFlag:是否包含balance属性
    • hintMap:注解类型和注解sql语句的map
    • 其他类似

对于路由模块,他需要完成的操作就是MyCat的核心功能之一,将前端发送过来的SQL语句路由到后面合适的分片上。那么,我们至少需要从SQL中解析出来这个SQL对应的是那张表,对应的分片规则是什么?有没有筛选条件,根据筛选条件我们是不是能路由到某几个分片上。是不是插入语句,需不需要生成全局唯一ID?等等等等
MyCat路由模块,大致上包括SQL语句分类,SQL语义解析,SQL语句改写,全局ID生成。

202401182020386423.png

5.2 SQL语句分类

首先,我们先回顾下,SQL语句通过客户端发送给了MyCat,MyCat在前端连接模块完成包解码,在这之后,对SQL语句进行分类处理(其实就是构建自己一套简单的语法分支)。
如何分类?其实就是通过语句第一个词先进行第一步分类:

202401182020389364.png

202401182020392935.png

ServerQueryHandler.java:

    public void query(String sql) {
    
            ServerConnection c = this.source;
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug(new StringBuilder().append(c).append(sql).toString());
            }
            //
            int rs = ServerParse.parse(sql);
            int sqlType = rs & 0xff;
    
            switch (sqlType) {
            //explain sql
            case ServerParse.EXPLAIN:
                ExplainHandler.handle(sql, c, rs >>> 8);
                break;
            //explain2 datanode=? sql=?
            case ServerParse.EXPLAIN2:
                Explain2Handler.handle(sql, c, rs >>> 8);
                break;
            case ServerParse.SET:
                SetHandler.handle(sql, c, rs >>> 8);
                break;
            case ServerParse.SHOW:
                ShowHandler.handle(sql, c, rs >>> 8);
                break;
            case ServerParse.SELECT:
                if(QuarantineHandler.handle(sql, c)){
                    SelectHandler.handle(sql, c, rs >>> 8);
                }
                break;
            case ServerParse.START:
                StartHandler.handle(sql, c, rs >>> 8);
                break;
            case ServerParse.BEGIN:
                BeginHandler.handle(sql, c);
                break;
            //不支持oracle的savepoint事务回退点
            case ServerParse.SAVEPOINT:
                SavepointHandler.handle(sql, c);
                break;
            case ServerParse.KILL:
                KillHandler.handle(sql, rs >>> 8, c);
                break;
            //不支持KILL_Query
            case ServerParse.KILL_QUERY:
                LOGGER.warn(new StringBuilder().append("Unsupported command:").append(sql).toString());
                c.writeErrMessage(ErrorCode.ER_UNKNOWN_COM_ERROR,"Unsupported command");
                break;
            case ServerParse.USE:
                UseHandler.handle(sql, c, rs >>> 8);
                break;
            case ServerParse.COMMIT:
                c.commit();
                break;
            case ServerParse.ROLLBACK:
                c.rollback();
                break;
            case ServerParse.HELP:
                LOGGER.warn(new StringBuilder().append("Unsupported command:").append(sql).toString());
                c.writeErrMessage(ErrorCode.ER_SYNTAX_ERROR, "Unsupported command");
                break;
            case ServerParse.MYSQL_CMD_COMMENT:
                c.write(c.writeToBuffer(OkPacket.OK, c.allocate()));
                break;
            case ServerParse.MYSQL_COMMENT:
                c.write(c.writeToBuffer(OkPacket.OK, c.allocate()));
                break;
                case ServerParse.LOAD_DATA_INFILE_SQL:
                    c.loadDataInfileStart(sql);
                    break;
            default:
                if(readOnly){
                    LOGGER.warn(new StringBuilder().append("User readonly:").append(sql).toString());
                    c.writeErrMessage(ErrorCode.ER_USER_READ_ONLY, "User readonly");
                    break;
                }
                if(QuarantineHandler.handle(sql, c)){
                    c.execute(sql, rs & 0xff);
                }
            }
        }

每种语句都有自己对应的Handler,我们这里将用Select语句举例。第一个词决定语句是什么类型(CURD)的,第二个词将更细粒度的区分语句,这里是不同的Select。首先我们思考下,不是所有的select语句都需要路由到后面数据库的。比如 select version这样的语句,可以直接回复MyCat的version。还有select LAST_INSERT_ID这样的(MyCat),上次插入的全局ID是在MyCat会缓存的。
所以,MyCat对于select的第二个词也做解析,可以将select语句分为可以直接回复的和必须路由到后面分片得到结果的。
SelectHandler.java:

    public static void handle(String stmt, ServerConnection c, int offs) {
            int offset = offs;
            switch (ServerParseSelect.parse(stmt, offs)) {
            case ServerParseSelect.VERSION_COMMENT:
                SelectVersionComment.response(c);
                break;
            case ServerParseSelect.DATABASE:
                SelectDatabase.response(c);
                break;
            case ServerParseSelect.USER:
                SelectUser.response(c);
                break;
            case ServerParseSelect.VERSION:
                SelectVersion.response(c);
                break;
            case ServerParseSelect.SESSION_INCREMENT:
                SessionIncrement.response(c);
                break;
            case ServerParseSelect.SESSION_ISOLATION:
                SessionIsolation.response(c);
                break;
            case ServerParseSelect.LAST_INSERT_ID:
                // offset = ParseUtil.move(stmt, 0, "select".length());
                loop:for (int l=stmt.length(); offset < l; ++offset) {
                    switch (stmt.charAt(offset)) {
                    case ' ':
                        continue;
                    case '/':
                    case '#':
                        offset = ParseUtil.comment(stmt, offset);
                        continue;
                    case 'L':
                    case 'l':
                        break loop;
                    }
                }
                offset = ServerParseSelect.indexAfterLastInsertIdFunc(stmt, offset);
                offset = ServerParseSelect.skipAs(stmt, offset);
                SelectLastInsertId.response(c, stmt, offset);
                break;
            case ServerParseSelect.IDENTITY:
                // offset = ParseUtil.move(stmt, 0, "select".length());
                loop:for (int l=stmt.length(); offset < l; ++offset) {
                    switch (stmt.charAt(offset)) {
                    case ' ':
                        continue;
                    case '/':
                    case '#':
                        offset = ParseUtil.comment(stmt, offset);
                        continue;
                    case '@':
                        break loop;
                    }
                }
                int indexOfAtAt = offset;
                offset += 2;
                offset = ServerParseSelect.indexAfterIdentity(stmt, offset);
                String orgName = stmt.substring(indexOfAtAt, offset);
                offset = ServerParseSelect.skipAs(stmt, offset);
                SelectIdentity.response(c, stmt, offset, orgName);
                break;
                case ServerParseSelect.SELECT_VAR_ALL:
                    SelectVariables.execute(c,stmt);
                    break;
            default:
                c.execute(stmt, ServerParse.SELECT);
            }
        }

202401182020395876.png
下一步,ServerConnection类处理SQL语句

202401182020399977.png

ServerConnection.java

    public void execute(String sql, int type) {
            //连接状态检查
            if (this.isClosed()) {
                LOGGER.warn("ignore execute ,server connection is closed " + this);
                return;
            }
            // 事务状态检查
            if (txInterrupted) {
                writeErrMessage(ErrorCode.ER_YES,
                        "Transaction error, need to rollback." + txInterrputMsg);
                return;
            }
    
            // 检查当前使用的DB
            String db = this.schema;
            if (db == null) {
                db = SchemaUtil.detectDefaultDb(sql, type);
                if (db == null) {
                    writeErrMessage(ErrorCode.ERR_BAD_LOGICDB, "No MyCAT Database selected");
                    return;
                }
            }
    
            // 兼容PhpAdmin's, 支持对MySQL元数据的模拟返回
             TODO: 2016/5/20 支持更多information_schema特性 
            if (ServerParse.SELECT == type 
                    && db.equalsIgnoreCase("information_schema") ) {
                MysqlInformationSchemaHandler.handle(sql, this);
                return;
            }
    
            if (ServerParse.SELECT == type 
                    && sql.contains("mysql") 
                    && sql.contains("proc")) {
    
                SchemaUtil.SchemaInfo schemaInfo = SchemaUtil.parseSchema(sql);
                if (schemaInfo != null 
                        && "mysql".equalsIgnoreCase(schemaInfo.schema)
                        && "proc".equalsIgnoreCase(schemaInfo.table)) {
    
                    // 兼容MySQLWorkbench
                    MysqlProcHandler.handle(sql, this);
                    return;
                }
            }
    
            SchemaConfig schema = MycatServer.getInstance().getConfig().getSchemas().get(db);
            if (schema == null) {
                writeErrMessage(ErrorCode.ERR_BAD_LOGICDB,
                        "Unknown MyCAT Database '" + db + "'");
                return;
            }
    
            routeEndExecuteSQL(sql, type, schema);
    
        }

调用routeEndExecuteSQL方法,会解析出RouteResultSet。这步包含了SQL语义解析,SQL路由,SQL查询优化,SQL语句改写,全局ID生成,最后,将解析出的RouteResultSet交给这个链接对应的session进行处理。
我们先分析SQL语义解析。看调用:
ServerConnection.java

    rrs = MycatServer
                        .getInstance()
                        .getRouterservice()
                        .route(MycatServer.getInstance().getConfig().getSystem(),
                                schema, type, sql, this.charset, this);

首先,关注下这个Routerservice是啥?在MyCat初始化时,会新建一个Routerservice(如之前配置模块中所讲):
MyCatServer.java

    //路由计算初始化
    routerService = new RouteService(cacheService);

Routerservice结构:

202401182020403568.png
其中sqlRouteCache和tableId2DataNodeCache是通过CacheService(MyCat里面是ehcache做的缓存)传入的对于sql语句缓存和tableid与后台分片对应关系的缓存。具体缓存会在缓存模块中讲。

调用route方法解析出RouteResultSet

202401182020406419.png

    public RouteResultset route(SystemConfig sysconf, SchemaConfig schema,
                int sqlType, String stmt, String charset, ServerConnection sc)
                throws SQLNonTransientException {
            RouteResultset rrs = null;
            String cacheKey = null;
    
            /**
             *  SELECT 类型的SQL, 检测
             */
            if (sqlType == ServerParse.SELECT) {
                cacheKey = schema.getName() + stmt;         
                rrs = (RouteResultset) sqlRouteCache.get(cacheKey);
                if (rrs != null) {
                    return rrs;
                }
            }
    
            /*!mycat: sql = select name from aa */
            /*!mycat: schema = test */
    //      boolean isMatchOldHint = stmt.startsWith(OLD_MYCAT_HINT);
    //      boolean isMatchNewHint = stmt.startsWith(NEW_MYCAT_HINT);
    //      if (isMatchOldHint || isMatchNewHint ) {
            int hintLength = RouteService.isHintSql(stmt);
            if(hintLength != -1){
                int endPos = stmt.indexOf("*/");
                if (endPos > 0) {               
                    // 用!mycat:内部的语句来做路由分析
    //              int hintLength = isMatchOldHint ? OLD_MYCAT_HINT.length() : NEW_MYCAT_HINT.length();
                    String hint = stmt.substring(hintLength, endPos).trim();    
    
                    int firstSplitPos = hint.indexOf(HINT_SPLIT);                
                    if(firstSplitPos > 0 ){
                        Map hintMap=    parseHint(hint);
                        String hintType = (String) hintMap.get(MYCAT_HINT_TYPE);
                        String hintSql = (String) hintMap.get(hintType);
                        if( hintSql.length() == 0 ) {
                            LOGGER.warn("comment int sql must meet :/*!mycat:type=value*/ or /*#mycat:type=value*/ or /*mycat:type=value*/: "+stmt);
                            throw new SQLSyntaxErrorException("comment int sql must meet :/*!mycat:type=value*/ or /*#mycat:type=value*/ or /*mycat:type=value*/: "+stmt);
                        }
                        String realSQL = stmt.substring(endPos + "*/".length()).trim();
    
                        HintHandler hintHandler = HintHandlerFactory.getHintHandler(hintType);
                        if( hintHandler != null ) {    
    
                            if ( hintHandler instanceof  HintSQLHandler) {                          
                                /**
                                 * 修复 注解SQL的 sqlType 与 实际SQL的 sqlType 不一致问题, 如: hint=SELECT,real=INSERT
                                 * fixed by zhuam
                                 */
                                int hintSqlType = ServerParse.parse( hintSql ) & 0xff;     
                                rrs = hintHandler.route(sysconf, schema, sqlType, realSQL, charset, sc, tableId2DataNodeCache, hintSql,hintSqlType,hintMap);
    
                            } else {                            
                                rrs = hintHandler.route(sysconf, schema, sqlType, realSQL, charset, sc, tableId2DataNodeCache, hintSql,sqlType,hintMap);
                            }
    
                        }else{
                            LOGGER.warn("TODO , support hint sql type : " + hintType);
                        }
    
                    }else{//fixed by runfriends@126.com
                        LOGGER.warn("comment in sql must meet :/*!mycat:type=value*/ or /*#mycat:type=value*/ or /*mycat:type=value*/: "+stmt);
                        throw new SQLSyntaxErrorException("comment in sql must meet :/*!mcat:type=value*/ or /*#mycat:type=value*/ or /*mycat:type=value*/: "+stmt);
                    }
                }
            } else {
                stmt = stmt.trim();
                rrs = RouteStrategyFactory.getRouteStrategy().route(sysconf, schema, sqlType, stmt,
                        charset, sc, tableId2DataNodeCache);
            }
    
            if (rrs != null && sqlType == ServerParse.SELECT && rrs.isCacheAble()) {
                sqlRouteCache.putIfAbsent(cacheKey, rrs);
            }
            return rrs;
        }

由于注解处理和sql解析有重叠,而且注解处理一直代码不稳定,所以,这里不涉及。只说sql正常解析的步骤

阅读全文
  • 点赞