【Mycat1.6之加载Schema的table 源码解读(一)】

gaojingsong

浏览: 1201737 次
性别:
来自: 深圳

最近访客更多访客>>

boveysmith

zah5897

xckouy

lengyun3566

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

博客分类：

Mycat中间件

Mycat1.6之加载Schema的table 源码解读

一、schema属性介绍

<table name="user" dataNode="dn1,dn2,dn3" rule="auto-sharding-long"

nameSuffix="" primaryKey="" autoIncrement="" needAddLimit="" type="" ruleRequired="" subTables=""/>

</schema>

primaryKey记录主键，用于之后路由分析，以及启用自增长主键

nameSuffix路由, 增加对动态日期表的支持

autoIncrement记录是否主键自增，默认不是，（启用全局sequence handler）

needAddLimit记录是否需要加返回结果集限制，默认需要加

type记录type，是否为global

ruleRequired记录是否绑定有分片规则

subTables分表功能

二、源码解读

private void load(String dtdFile, String xmlFile) {
	InputStream dtd = null;
	InputStream xml = null; 
	dtd = XMLSchemaLoader.class.getResourceAsStream(dtdFile);
	xml = XMLSchemaLoader.class.getResourceAsStream(xmlFile);
	Element root = ConfigUtil.getDocument(dtd, xml).getDocumentElement();
	//先加载所有的DataHost
	loadDataHosts(root);
	//再加载所有的DataNode
	loadDataNodes(root);
	//最后加载所有的Schema
	loadSchemas(root);	 
}

<schema name="TESTDB" checkSQLschema="false" sqlMaxLimit="100" dataNode="">
	<table name="tab" dataNode="dn1,dn2,dn3" rule="auto-sharding-long" />
	<table name="user1" dataNode="dn1,dn2,dn3" rule="auto-sharding-long" />
	<table name="user"  dataNode="dn1,dn2,dn3" rule="auto-sharding-long" />
</schema>
private void loadSchemas(Element root) {
	NodeList list = root.getElementsByTagName("schema");
	for (int i = 0, n = list.getLength(); i < n; i++) {
		Element schemaElement = (Element) list.item(i);
		//读取各个属性
		String name = schemaElement.getAttribute("name");
		String dataNode = schemaElement.getAttribute("dataNode");
		String checkSQLSchemaStr = schemaElement.getAttribute("checkSQLschema");
		String sqlMaxLimitStr = schemaElement.getAttribute("sqlMaxLimit");
		int sqlMaxLimit = -1;
		//读取sql返回结果集限制
		if (sqlMaxLimitStr != null && !sqlMaxLimitStr.isEmpty()) {
			sqlMaxLimit = Integer.parseInt(sqlMaxLimitStr);
		}
		
		// check dataNode already exists or not,看schema标签中是否有datanode
		String defaultDbType = null;
		//校验检查并添加dataNode
		if (dataNode != null && !dataNode.isEmpty()) {
			List dataNodeLst = new ArrayList(1);
			dataNodeLst.add(dataNode);
			checkDataNodeExists(dataNodeLst);
			String dataHost = dataNodes.get(dataNode).getDataHost();
			defaultDbType = dataHosts.get(dataHost).getDbType();
		} else {
			dataNode = null;
		}
		//加载schema下所有tables
		Map<String, TableConfig> tables = loadTables(schemaElement);
		//判断schema是否重复
		if (schemas.containsKey(name)) {
			throw new ConfigException("schema " + name + " duplicated!");
		}

		// 设置了table的不需要设置dataNode属性，没有设置table的必须设置dataNode属性
		if (dataNode == null && tables.size() == 0) {
			throw new ConfigException(
					"schema " + name + " didn't config tables,so you must set dataNode property!");
		}

		SchemaConfig schemaConfig = new SchemaConfig(name, dataNode,
				tables, sqlMaxLimit, "true".equalsIgnoreCase(checkSQLSchemaStr));

		//设定DB类型，这对之后的sql语句路由解析有帮助
		if (defaultDbType != null) {
			schemaConfig.setDefaultDataNodeDbType(defaultDbType);
			if (!"mysql".equalsIgnoreCase(defaultDbType)) {
				schemaConfig.setNeedSupportMultiDBType(true);
			}
		}

		// 判断是否有不是mysql的数据库类型，方便解析判断是否启用多数据库分页语法解析
		for (TableConfig tableConfig : tables.values()) {
			if (isHasMultiDbType(tableConfig)) {
				schemaConfig.setNeedSupportMultiDBType(true);
				break;
			}
		}
		//记录每种dataNode的DB类型
		Map<String, String> dataNodeDbTypeMap = new HashMap<>();
		for (String dataNodeName : dataNodes.keySet()) {
			DataNodeConfig dataNodeConfig = dataNodes.get(dataNodeName);
			String dataHost = dataNodeConfig.getDataHost();
			DataHostConfig dataHostConfig = dataHosts.get(dataHost);
			if (dataHostConfig != null) {
				String dbType = dataHostConfig.getDbType();
				dataNodeDbTypeMap.put(dataNodeName, dbType);
			}
		}
		schemaConfig.setDataNodeDbTypeMap(dataNodeDbTypeMap);
		schemas.put(name, schemaConfig);
	}
}

private Map<String, TableConfig> loadTables(Element node) {
	// Map<String, TableConfig> tables = new HashMap<String, TableConfig>();
	// 支持表名中包含引号[`] BEN GONG
	Map<String, TableConfig> tables = new TableConfigMap();
	NodeList nodeList = node.getElementsByTagName("table");
	for (int i = 0; i < nodeList.getLength(); i++) {
		Element tableElement = (Element) nodeList.item(i);
		String tableNameElement = tableElement.getAttribute("name").toUpperCase();

		//TODO:路由, 增加对动态日期表的支持
		String tableNameSuffixElement = tableElement.getAttribute("nameSuffix").toUpperCase();
		if ( !"".equals( tableNameSuffixElement ) ) {				
			
			if( tableNameElement.split(",").length > 1 ) {
				throw new ConfigException("nameSuffix " + 
tableNameSuffixElement + ", require name parameter cannot multiple breaks!");
			}
			//前缀用来标明日期格式
			tableNameElement = doTableNameSuffix(tableNameElement, tableNameSuffixElement);
		}
		//记录主键，用于之后路由分析，以及启用自增长主键
		String[] tableNames = tableNameElement.split(",");
		String primaryKey = tableElement.hasAttribute("primaryKey") ? 
tableElement.getAttribute("primaryKey").toUpperCase() : null;
		//记录是否主键自增，默认不是，（启用全局sequence handler）
		boolean autoIncrement = false;
		if (tableElement.hasAttribute("autoIncrement")) {
			autoIncrement = Boolean.parseBoolean(tableElement.getAttribute("autoIncrement"));
		}
		//记录是否需要加返回结果集限制，默认需要加
		boolean needAddLimit = true;
		if (tableElement.hasAttribute("needAddLimit")) {
			needAddLimit = Boolean.parseBoolean(tableElement.getAttribute("needAddLimit"));
		}
		//记录type，是否为global
		String tableTypeStr = tableElement.hasAttribute("type") ? 
tableElement.getAttribute("type") : null;
		int tableType = TableConfig.TYPE_GLOBAL_DEFAULT;
		if ("global".equalsIgnoreCase(tableTypeStr)) {
			tableType = TableConfig.TYPE_GLOBAL_TABLE;
		}
		//记录dataNode，就是分布在哪些dataNode上
		String dataNode = tableElement.getAttribute("dataNode");
		TableRuleConfig tableRule = null;
		if (tableElement.hasAttribute("rule")) {
			String ruleName = tableElement.getAttribute("rule");
			tableRule = tableRules.get(ruleName);
			if (tableRule == null) {
				throw new ConfigException("rule " + ruleName + " is not found!");
			}
		}
		
		boolean ruleRequired = false;
		//记录是否绑定有分片规则
		if (tableElement.hasAttribute("ruleRequired")) {
			ruleRequired = Boolean.parseBoolean(tableElement.getAttribute("ruleRequired"));
		}

		if (tableNames == null) {
			throw new ConfigException("table name is not found!");
		}
		//distribute函数，重新编排dataNode
		String distPrex = "distribute(";
		boolean distTableDns = dataNode.startsWith(distPrex);
		if (distTableDns) {
			dataNode = dataNode.substring(distPrex.length(), dataNode.length() - 1);
		}
		//分表功能
		String subTables = tableElement.getAttribute("subTables");
		
		for (int j = 0; j < tableNames.length; j++) {

			String tableName = tableNames[j];
			TableRuleConfig	tableRuleConfig=tableRule ;
			  if(tableRuleConfig!=null) {
				//对于实现TableRuleAware的function进行特殊处理  根据每个表新建个实例
				  RuleConfig rule= tableRuleConfig.getRule();
				  if(rule.getRuleAlgorithm() instanceof TableRuleAware)  {
					  tableRuleConfig = (TableRuleConfig) ObjectUtil.copyObject(tableRuleConfig);
					  tableRules.remove(tableRuleConfig.getName())   ;
					  String newRuleName = tableRuleConfig.getName() + "_" + tableName;
					  tableRuleConfig. setName(newRuleName);
					  TableRuleAware tableRuleAware= (TableRuleAware) 
tableRuleConfig.getRule().getRuleAlgorithm();
					  tableRuleAware.setRuleName(newRuleName);
					  tableRuleAware.setTableName(tableName);
					  tableRuleConfig.getRule().getRuleAlgorithm().init();
					  tableRules.put(newRuleName,tableRuleConfig);
				  }
			  }

			TableConfig table = new TableConfig(tableName, primaryKey,
					autoIncrement, needAddLimit, tableType, dataNode,
					getDbType(dataNode),
					(tableRuleConfig != null) ? tableRuleConfig.getRule() : null,
					ruleRequired, null, false, null, null,subTables);
			
			checkDataNodeExists(table.getDataNodes());
			// 检查分片表分片规则配置是否合法
			if(table.getRule() != null) {
				checkRuleSuitTable(table);
			}
			
			if (distTableDns) {
				distributeDataNodes(table.getDataNodes());
			}
			//检查去重
			if (tables.containsKey(table.getName())) {
				throw new ConfigException("table " + tableName + " duplicated!");
			}
			//放入map
			tables.put(table.getName(), table);
		}
		//只有tableName配置的是单个表（没有逗号）的时候才能有子表
		if (tableNames.length == 1) {
			TableConfig table = tables.get(tableNames[0]);
			// process child tables  递归调用
			processChildTables(tables, table, dataNode, tableElement);
		}
	}
	return tables;
}
	
	
/**
* 处理动态日期表, 支持 YYYYMM、YYYYMMDD 两种格式
* 
* YYYYMM格式： 	  yyyymm,2015,01,60   
* YYYYMMDD格式:  yyyymmdd,2015,01,10,50
* 
* @param tableNameElement
* @param tableNameSuffixElement
 * @return
 */
private String doTableNameSuffix(String tableNameElement, String tableNameSuffixElement) {
		
	String newTableName = tableNameElement;
	
	String[] params = tableNameSuffixElement.split(",");			
	String suffixFormat = params[0].toUpperCase();		
	if ( suffixFormat.equals("YYYYMM") ) {
		
		//读取参数
		int yyyy = Integer.parseInt( params[1] );
		int mm = Integer.parseInt( params[2] );					
		int mmEndIdx =  Integer.parseInt( params[3] );
		
		//日期处理
		SimpleDateFormat yyyyMMSDF = new SimpleDateFormat("yyyyMM"); 
		
		Calendar cal = Calendar.getInstance();
		cal.set(Calendar.YEAR, yyyy );
		cal.set(Calendar.MONTH, mm - 1 );
		cal.set(Calendar.DATE, 0 );
		
		//表名改写
		StringBuffer tableNameBuffer = new StringBuffer();
		for(int mmIdx = 0; mmIdx <= mmEndIdx; mmIdx++) {						
			tableNameBuffer.append( tableNameElement );
			tableNameBuffer.append( yyyyMMSDF.format(cal.getTime()) );							
			cal.add(Calendar.MONTH, 1);
			
			if ( mmIdx != mmEndIdx) {
				tableNameBuffer.append(",");
			}						
		}
                //逗号分隔的一批表名字A1,B2,C3					
		newTableName = tableNameBuffer.toString();

	} else if ( suffixFormat.equals("YYYYMMDD") ) {
		
		//读取参数
		int yyyy = Integer.parseInt( params[1] );
		int mm = Integer.parseInt( params[2] );
		int dd =  Integer.parseInt( params[3] );
		int ddEndIdx =  Integer.parseInt( params[4] );
		
		//日期处理
		SimpleDateFormat yyyyMMddSDF = new SimpleDateFormat("yyyyMMdd"); 
		
		Calendar cal = Calendar.getInstance();
		cal.set(Calendar.YEAR, yyyy );
		cal.set(Calendar.MONTH, mm - 1 );
		cal.set(Calendar.DATE, dd );
		
		//表名改写
		StringBuffer tableNameBuffer = new StringBuffer();
		for(int ddIdx = 0; ddIdx <= ddEndIdx; ddIdx++) {					
			tableNameBuffer.append( tableNameElement );
			tableNameBuffer.append( yyyyMMddSDF.format(cal.getTime()) );
			
			cal.add(Calendar.DATE, 1);	
			
			if ( ddIdx != ddEndIdx) {
				tableNameBuffer.append(",");
			}
		}
               //逗号分隔的一批表名字A1,B2,C3					
		newTableName = tableNameBuffer.toString();
	}				
	return newTableName;		
}

三、动态日期表演示

</schema>

查看图片附件

0
顶

1
踩

分享到：

【elasticsearch建立索引操作的API】 | 【编程语言之Erlang介绍】

2017-05-16 23:29
浏览 1362
评论(0)
分类:开源软件
查看更多

发表评论

您还没有登录,请您登录后再发表评论

最近访客更多访客>>

博主相关

文章分类

社区版块

存档分类

最新评论