【Mycat1.6之加载Schema的DataHosts完整属性源码解读】

一、Mycat1.6只Schema加载DataHosts完整属性源码解读

        private final static String DEFAULT_DTD = "/schema.dtd";
	private final static String DEFAULT_XML = "/schema.xml";

	public XMLSchemaLoader(String schemaFile, String ruleFile) {
		//先读取rule.xml
		XMLRuleLoader ruleLoader = new XMLRuleLoader(ruleFile);
		//将tableRules拿出，用于这里加载Schema做rule有效判断，以及之后的分片路由计算
		this.tableRules = ruleLoader.getTableRules();
		//释放ruleLoader
		ruleLoader = null;
		this.dataHosts = new HashMap<String, DataHostConfig>();
		this.dataNodes = new HashMap<String, DataNodeConfig>();
		this.schemas = new HashMap<String, SchemaConfig>();
		//读取加载schema配置
		this.load(DEFAULT_DTD, schemaFile == null ? DEFAULT_XML : schemaFile);
	}
	
	//加载顺序：主机/节点/ Schema
	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);
		 
	}
	 
       /**
<dataHost name="localhost1_gaojingsong1" maxCon="1000" minCon="1" balance="0" writeType="0" 
dbType="db2" dbDriver="jdbc" switchType="-1" slaveThreshold="" tempReadHostAvailable=""  
filters="" logTime="">
<heartbeat>select 1 from sysibm.sysdummy1</heartbeat>
<connectionInitSql>alter session set nls_date_format='yyyy-mm-dd hh24:mi:ss'
</connectionInitSql>
<writeHost host="hostM1" url="jdbc:db2://19:50000/TEST001" user="db2inst1" password="123456" >
</writeHost> 
</dataHost>
	*/
	private void loadDataHosts(Element root) {
		NodeList list = root.getElementsByTagName("dataHost");
		for (int i = 0, n = list.getLength(); i < n; ++i) {
			
			Element element = (Element) list.item(i);
			String name = element.getAttribute("name");
			//判断是否重复
			if (dataHosts.containsKey(name)) {
				throw new ConfigException("dataHost name " + name + "duplicated!");
			}
			//读取最大连接数
			int maxCon = Integer.parseInt(element.getAttribute("maxCon"));
			//读取最小连接数
			int minCon = Integer.parseInt(element.getAttribute("minCon"));
	/**
	* 读取负载均衡配置
	* 1. balance="0", 不开启分离机制，所有读操作都发送到当前可用的 writeHost 上。
	* 2. balance="1"，全部的 readHost 和 stand by writeHost 参不 select 的负载均衡
	* 3. balance="2"，所有读操作都随机的在 writeHost、readhost 上分发。
	* 4. balance="3"，所有读请求随机的分发到 wiriterHost 对应的 readhost 执行，writerHost 不负担读压力
	 */
			int balance = Integer.parseInt(element.getAttribute("balance"));
			/**
			 * 读取切换类型
			 * -1 表示不自动切换
			 * 1 默认值，自动切换
			 * 2 基于MySQL主从同步的状态决定是否切换
			 * 心跳询句为 show slave status
			 * 3 基于 MySQL galary cluster 的切换机制
			 */
			String switchTypeStr = element.getAttribute("switchType");
	int switchType = switchTypeStr.equals("") ? -1 :Integer.parseInt(switchTypeStr);
			//读取从延迟界限
			String slaveThresholdStr = element.getAttribute("slaveThreshold");
	int slaveThreshold = slaveThresholdStr.equals("") ? -1 : Integer.parseInt(slaveThresholdStr);
			
    //如果 tempReadHostAvailable 设置大于 0 则表示写主机如果挂掉，临时的读服务依然可用
     String tempReadHostAvailableStr = element.getAttribute("tempReadHostAvailable");
     boolean tempReadHostAvailable = !tempReadHostAvailableStr.equals("")
 && Integer.parseInt(tempReadHostAvailableStr) > 0;
			/**
			 * 读取 写类型
			 * 这里只支持 0 - 所有写操作仅配置的第一个 writeHost
			 */
			String writeTypStr = element.getAttribute("writeType");
int writeType = "".equals(writeTypStr) ? PhysicalDBPool.WRITE_ONLYONE_NODE : Integer.parseInt(writeTypStr);


			String dbDriver = element.getAttribute("dbDriver");
			String dbType = element.getAttribute("dbType");
			String filters = element.getAttribute("filters");
			String logTimeStr = element.getAttribute("logTime");
long logTime = "".equals(logTimeStr) ? PhysicalDBPool.LONG_TIME : Long.parseLong(logTimeStr) ;
//读取心跳语句
String heartbeatSQL = element.getElementsByTagName("heartbeat").item(0).getTextContent();
//读取 初始化sql配置,用于oracle
NodeList connectionInitSqlList = element.getElementsByTagName("connectionInitSql");
			String initConSQL = null;
			if (connectionInitSqlList.getLength() > 0) {
				initConSQL = connectionInitSqlList.item(0).getTextContent();
			}
	//读取writeHost
	/**
	<writeHost host="hostM1" url="localhost:3306" user="root"
		 password="123456" usingDecrypt="" weight="">
	<!-- can have multi read hosts -->
	<readHost host="hostS2" url="192.168.1.200:3306" user="root" password="xxx" />
	</writeHost>
	*/
	NodeList writeNodes = element.getElementsByTagName("writeHost");
	DBHostConfig[] writeDbConfs = new DBHostConfig[writeNodes.getLength()];
	Map<Integer, DBHostConfig[]> readHostsMap = new HashMap<Integer, DBHostConfig[]>(2);
			for (int w = 0; w < writeDbConfs.length; w++) {
				Element writeNode = (Element) writeNodes.item(w);
writeDbConfs[w] = createDBHostConf(name, writeNode, dbType,dbDriver, maxCon, minCon,filters,logTime);
				NodeList readNodes = writeNode.getElementsByTagName("readHost");
				//读取对应的每一个readHost
				if (readNodes.getLength() != 0) {
DBHostConfig[] readDbConfs = new DBHostConfig[readNodes.getLength()];
					for (int r = 0; r < readDbConfs.length; r++) {
						Element readNode = (Element) readNodes.item(r);
readDbConfs[r] = createDBHostConf(name,readNode, dbType, dbDriver, maxCon, minCon,filters, logTime);
					}
					readHostsMap.put(w, readDbConfs);
				}
			}

DataHostConfig hostConf = new DataHostConfig(name, dbType, dbDriver, writeDbConfs, 
readHostsMap, switchType, slaveThreshold, tempReadHostAvailable);		
			
			hostConf.setMaxCon(maxCon);
			hostConf.setMinCon(minCon);
			hostConf.setBalance(balance);
			hostConf.setWriteType(writeType);
			hostConf.setHearbeatSQL(heartbeatSQL);
			hostConf.setConnectionInitSql(initConSQL);
			hostConf.setFilters(filters);
			hostConf.setLogTime(logTime);
			dataHosts.put(hostConf.getName(), hostConf);
		}
	}
	
	private DBHostConfig createDBHostConf(String dataHost, Element node,
String dbType, String dbDriver, int maxCon, int minCon, 
String filters, long logTime) {
		
		String nodeHost = node.getAttribute("host");
		String nodeUrl = node.getAttribute("url");
		String user = node.getAttribute("user");
		String password = node.getAttribute("password");
		String usingDecrypt = node.getAttribute("usingDecrypt");
		String passwordEncryty= DecryptUtil.DBHostDecrypt(usingDecrypt, 
nodeHost, user, password);
		
		String weightStr = node.getAttribute("weight");
		int weight = "".equals(weightStr) ? PhysicalDBPool.WEIGHT :
 Integer.parseInt(weightStr) ;
		
		String ip = null;
		int port = 0;
		if (empty(nodeHost) || empty(nodeUrl) || empty(user)) {
			throw new ConfigException(
					"dataHost "
							+ dataHost
		+ " define error,some attributes of this element is empty: "
							+ nodeHost);
		}
		if ("native".equalsIgnoreCase(dbDriver)) {
			int colonIndex = nodeUrl.indexOf(':');
			ip = nodeUrl.substring(0, colonIndex).trim();
			port = Integer.parseInt(nodeUrl.substring(colonIndex + 1).trim());
		} else {
			URI url;
			try {
				url = new URI(nodeUrl.substring(5));
			} catch (Exception e) {
				throw new ConfigException("invalid jdbc url " + nodeUrl + 
" of " + dataHost);
			}
			ip = url.getHost();
			port = url.getPort();
		}

		DBHostConfig conf = new DBHostConfig(nodeHost, ip, port, nodeUrl, user,
 passwordEncryty,password);
		conf.setDbType(dbType);
		conf.setMaxCon(maxCon);
		conf.setMinCon(minCon);
		conf.setFilters(filters);
		conf.setLogTime(logTime);
		conf.setWeight(weight); 	//新增权重
		return conf;
	}
	
	public static String DBHostDecrypt(String usingDecrypt,String host,
String user ,String passwrod){
		if("1".equals(usingDecrypt)){
			//type:host:user:password
        	//1:myhost1:test:test
        	boolean flag = false;
        	try {
        		String passwrods[] = DecryptUtil.decrypt(passwrod).split(":");
            	if("1".equals(passwrods[0]) && host.equals(passwrods[1]) && 
user.equals(passwrods[2])){
            		return passwrods[3];
            	}
            	if(flag==false){
            		 throw new ConfigException("user " + user + " passwrod need to decrype ,but decrype password is wrong !");
            	}
        	} catch (Exception e2) {
       		    throw new ConfigException("host " + host + ",user " + user + " passwrod need to decrype ,but decrype password is wrong !",e2);
			}
		}
		return passwrod;
	}

二、总结：

dataHost 完整配置如下：

/**

<dataHost name="localhost1_gaojingsong1" maxCon="1000" minCon="1" balance="0" writeType="0" 

dbType="db2" dbDriver="jdbc" switchType="-1" slaveThreshold="" tempReadHostAvailable=""  filters="" logTime="">

<heartbeat>select 1 from sysibm.sysdummy1</heartbeat>

<connectionInitSql>alter session set nls_date_format='yyyy-mm-dd hh24:mi:ss'</connectionInitSql>

<writeHost host="hostM1" url="jdbc:db2://192.168.8.129:50000/TEST001" user="db2inst1" password="123456" >

</writeHost> 

</dataHost>

*/

 

 

//读取writeHost

/**

<writeHost host="hostM1" url="localhost:3306" user="root"

 password="123456" usingDecrypt="" weight="">

<!-- can have multi read hosts -->

<readHost host="hostS2" url="192.168.1.200:3306" user="root" password="xxx"   usingDecrypt="" weight=""/>

</writeHost>

*/

 

备注：

writeType 这里只支持 0 - 所有写操作仅配置的第一个 writeHost

/** * balance读取负载均衡配置

* 1. balance="0", 不开启分离机制，所有读操作都发送到当前可用的 writeHost 上。

* 2. balance="1"，全部的 readHost 和 stand by writeHost 参不 select 的负载均衡

* 3. balance="2"，所有读操作都随机的在 writeHost、readhost 上分发。

* 4. balance="3"，所有读请求随机的分发到 wiriterHost 对应的 readhost 执行，

writerHost 不负担读压力

*/

 

/**

* switchType读取切换类型

* -1 表示不自动切换 * 1 默认值，自动切换

* 2 基于MySQL主从同步的状态决定是否切换

* 心跳询句为 show slave status

* 3 基于 MySQL galary cluster 的切换机制

*/

 

tempReadHostAvailable 设置大于 0 则表示写主机如果挂掉， 临时的读服务依然可用