5. 路由模块
5.1 路由模块组成：
路由模块，我们可以先把他当做个黑盒，看下输入和输出都是神马。
输入，很明显，就是个SQL语句，原生的，不加任何修饰的，纯洁的，从客户端发过来刚刚被解码的SQL语句。
输出呢？就是个优化，改写后的SQL语句，以及要发送到的后台分片。
这个RouteResultSet就是输出，长什么样子呢？
下图是主要涉及到的类：
-
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生成。
5.2 SQL语句分类
首先，我们先回顾下，SQL语句通过客户端发送给了MyCat，MyCat在前端连接模块完成包解码，在这之后，对SQL语句进行分类处理（其实就是构建自己一套简单的语法分支）。
如何分类？其实就是通过语句第一个词先进行第一步分类：
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);
}
}
下一步，ServerConnection类处理SQL语句
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结构：
其中sqlRouteCache和tableId2DataNodeCache是通过CacheService（MyCat里面是ehcache做的缓存）传入的对于sql语句缓存和tableid与后台分片对应关系的缓存。具体缓存会在缓存模块中讲。
调用route方法解析出RouteResultSet
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正常解析的步骤
