Mycat-Server-java入门

package io.mycat.route.util; import java.sql.SQLNonTransientException; import java.sql.SQLSyntaxErrorException; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; import java.util.LinkedHashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.Callable; import org.apache.commons.lang.StringUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import com.alibaba.druid.sql.ast.SQLExpr; import com.alibaba.druid.sql.ast.SQLStatement; import com.alibaba.druid.sql.ast.expr.SQLCharExpr; import com.alibaba.druid.sql.ast.expr.SQLIdentifierExpr; import com.alibaba.druid.sql.ast.statement.SQLCharacterDataType; import com.alibaba.druid.sql.ast.statement.SQLColumnDefinition; import com.alibaba.druid.sql.ast.statement.SQLCreateTableStatement; import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlCreateTableStatement; import com.alibaba.druid.sql.dialect.mysql.ast.statement.MySqlInsertStatement; import com.alibaba.druid.sql.dialect.mysql.parser.MySqlStatementParser; import com.alibaba.druid.wall.spi.WallVisitorUtils; import com.google.common.base.Strings; import com.google.common.collect.Maps; import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import io.mycat.MycatServer; import io.mycat.backend.datasource.PhysicalDBNode; import io.mycat.backend.datasource.PhysicalDBPool; import io.mycat.backend.datasource.PhysicalDatasource; import io.mycat.backend.mysql.nio.handler.FetchStoreNodeOfChildTableHandler; import io.mycat.backend.mysql.nio.handler.JDBCFetchStoreNodeOfChildTableHandler; import io.mycat.cache.LayerCachePool; import io.mycat.config.ErrorCode; import io.mycat.config.MycatConfig; import io.mycat.config.model.SchemaConfig; import io.mycat.config.model.TableConfig; import io.mycat.config.model.rule.RuleConfig; import io.mycat.route.RouteResultset; import io.mycat.route.RouteResultsetNode; import io.mycat.route.SessionSQLPair; import io.mycat.route.function.AbstractPartitionAlgorithm; import io.mycat.route.function.SlotFunction; import io.mycat.route.parser.druid.DruidShardingParseInfo; import io.mycat.route.parser.druid.RouteCalculateUnit; import io.mycat.server.ServerConnection; import io.mycat.server.parser.ServerParse; import io.mycat.sqlengine.mpp.ColumnRoutePair; import io.mycat.sqlengine.mpp.LoadData; import io.mycat.util.StringUtil; /** * 从ServerRouterUtil中抽取的一些公用方法，路由解析工具类 * @author wang.dw * */ public class RouterUtil { private static final Logger LOGGER = LoggerFactory.getLogger(RouterUtil.class); /** * 移除执行语句中的数据库名 * * @param stmt 执行语句 * @param schema 数据库名 * @return 执行语句 * @author mycat * * @modification 修正移除schema的方法 * @date 2016/12/29 * @modifiedBy Hash Zhang * */ public static String removeSchema(String stmt, String schema) { final String upStmt = stmt.toUpperCase(); final String upSchema = schema.toUpperCase() + "."; final String upSchema2 = new StringBuilder("`").append(schema.toUpperCase()).append("`.").toString(); int strtPos = 0; int indx = 0; int indx1 = upStmt.indexOf(upSchema, strtPos); int indx2 = upStmt.indexOf(upSchema2, strtPos); boolean flag = indx1 < indx2 ? indx1 == -1 : indx2 != -1; indx = !flag ? indx1 > 0 ? indx1 : indx2 : indx2 > 0 ? indx2 : indx1; if (indx < 0) { return stmt; } int firstE = upStmt.indexOf("'"); int endE = upStmt.lastIndexOf("'"); StringBuilder sb = new StringBuilder(); while (indx > 0) { sb.append(stmt.substring(strtPos, indx)); if (flag) { strtPos = indx + upSchema2.length(); } else { strtPos = indx + upSchema.length(); } if (indx > firstE && indx < endE && countChar(stmt, indx) % 2 == 1) { sb.append(stmt.substring(indx, indx + schema.length() + 1)); } indx1 = upStmt.indexOf(upSchema, strtPos); indx2 = upStmt.indexOf(upSchema2, strtPos); flag = indx1 < indx2 ? indx1 == -1 : indx2 != -1; indx = !flag ? indx1 > 0 ? indx1 : indx2 : indx2 > 0 ? indx2 : indx1; } sb.append(stmt.substring(strtPos)); return sb.toString(); } private static int countChar(String sql,int end) { int count=0; boolean skipChar = false; for (int i = 0; i < end; i++) { if(sql.charAt(i)=='\'' && !skipChar) { count++; skipChar = false; }else if( sql.charAt(i)=='\\'){ skipChar = true; }else{ skipChar = false; } } return count; } /** * 获取第一个节点作为路由 * * @param rrs 数据路由集合 * @param dataNode 数据库所在节点 * @param stmt 执行语句 * @return 数据路由集合 * * @author mycat */ public static RouteResultset routeToSingleNode(RouteResultset rrs, String dataNode, String stmt) { if (dataNode == null) { return rrs; } RouteResultsetNode[] nodes = new RouteResultsetNode[1]; nodes[0] = new RouteResultsetNode(dataNode, rrs.getSqlType(), stmt);//rrs.getStatement() nodes[0].setSource(rrs); rrs.setNodes(nodes); rrs.setFinishedRoute(true); if(rrs.getDataNodeSlotMap().containsKey(dataNode)){ nodes[0].setSlot(rrs.getDataNodeSlotMap().get(dataNode)); } if (rrs.getCanRunInReadDB() != null) { nodes[0].setCanRunInReadDB(rrs.getCanRunInReadDB()); } if(rrs.getRunOnSlave() != null){ nodes[0].setRunOnSlave(rrs.getRunOnSlave()); } return rrs; } /** * 修复DDL路由 * * @return RouteResultset * @author aStoneGod */ public static RouteResultset routeToDDLNode(RouteResultset rrs, int sqlType, String stmt,SchemaConfig schema) throws SQLSyntaxErrorException { stmt = getFixedSql(stmt); String tablename = ""; //去除sql前面的注册,如/* ApplicationName=DBeaver 6.0.1 - Main */，这个注册会导致create table出错 stmt = stmt.replaceFirst("\\/\\*.*\\*\\/\\s*", ""); final String upStmt = stmt.toUpperCase(); String rrsStmt = new String(upStmt); if(upStmt.startsWith("CREATE")){ if (upStmt.contains("CREATE INDEX ") || upStmt.contains("CREATE UNIQUE INDEX ")){ tablename = RouterUtil.getTableName(stmt, RouterUtil.getCreateIndexPos(upStmt, 0)); /** * Date：2017年11月2日 * @author SvenAugustus 修复oracle 语法不支持 drop index i_t_f on t_test，只有drop index i_t_f; */ if(!"oracle".equalsIgnoreCase(schema.getDefaultDataNodeDbType())){ int onInd = upStmt.indexOf("ON", 0); rrsStmt= upStmt.substring(0, onInd); } }else { tablename = RouterUtil.getTableName(stmt, RouterUtil.getCreateTablePos(upStmt, 0)); } }else if(upStmt.startsWith("DROP")){ if (upStmt.contains("DROP INDEX ")){ tablename = RouterUtil.getTableName(stmt, RouterUtil.getDropIndexPos(upStmt, 0)); }else { tablename = RouterUtil.getTableName(stmt, RouterUtil.getDropTablePos(upStmt, 0)); } }else if(upStmt.startsWith("ALTER")){ tablename = RouterUtil.getTableName(stmt, RouterUtil.getAlterTablePos(upStmt, 0)); }else if (upStmt.startsWith("TRUNCATE")){ tablename = RouterUtil.getTableName(stmt, RouterUtil.getTruncateTablePos(upStmt, 0)); } tablename = tablename.toUpperCase(); if (schema.getTables().containsKey(tablename)){ if(ServerParse.DDL==sqlType){ List<String> dataNodes = new ArrayList<>(); Map<String, TableConfig> tables = schema.getTables(); TableConfig tc=tables.get(tablename); if (tables != null && (tc != null)) { dataNodes = tc.getDataNodes(); } boolean isSlotFunction= tc.getRule() != null && t