abstractqueryrouter.java

来自「mysql集群」· Java 代码 · 共 963 行 · 第 1/3 页

/*
 * 	This program is free software; you can redistribute it and/or modify it under the terms of 
 * the GNU General Public License as published by the Free Software Foundation; either version 3 of the License, 
 * or (at your option) any later version. 
 * 
 * 	This program is distributed in the hope that it will be useful, 
 * but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  
 * See the GNU General Public License for more details. 
 * 	You should have received a copy of the GNU General Public License along with this program; 
 * if not, write to the Free Software Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
 */
package com.meidusa.amoeba.route;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import javax.xml.parsers.DocumentBuilder;
import javax.xml.parsers.DocumentBuilderFactory;

import org.apache.commons.collections.map.LRUMap;
import org.apache.log4j.Logger;
import org.apache.log4j.helpers.LogLog;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.Node;
import org.w3c.dom.NodeList;
import org.xml.sax.EntityResolver;
import org.xml.sax.ErrorHandler;
import org.xml.sax.InputSource;
import org.xml.sax.SAXException;
import org.xml.sax.SAXParseException;

import com.meidusa.amoeba.parser.dbobject.Column;
import com.meidusa.amoeba.parser.statment.CommitStatment;
import com.meidusa.amoeba.parser.statment.DMLStatment;
import com.meidusa.amoeba.parser.statment.PropertyStatment;
import com.meidusa.amoeba.parser.statment.RollbackStatment;
import com.meidusa.amoeba.parser.statment.StartTansactionStatment;
import com.meidusa.amoeba.parser.function.Function;
import com.meidusa.amoeba.parser.dbobject.Schema;
import com.meidusa.amoeba.parser.statment.Statment;
import com.meidusa.amoeba.parser.dbobject.Table;
import com.meidusa.amoeba.parser.expression.Expression;
import com.meidusa.amoeba.sqljep.function.Abs;
import com.meidusa.amoeba.sqljep.function.AddDate;
import com.meidusa.amoeba.sqljep.function.AddMonths;
import com.meidusa.amoeba.sqljep.function.AddTime;
import com.meidusa.amoeba.sqljep.function.Case;
import com.meidusa.amoeba.sqljep.function.Ceil;
import com.meidusa.amoeba.sqljep.function.Comparative;
import com.meidusa.amoeba.sqljep.function.ComparativeBaseList;
import com.meidusa.amoeba.sqljep.function.Concat;
import com.meidusa.amoeba.sqljep.function.Datediff;
import com.meidusa.amoeba.sqljep.function.Day;
import com.meidusa.amoeba.sqljep.function.DayName;
import com.meidusa.amoeba.sqljep.function.DayOfWeek;
import com.meidusa.amoeba.sqljep.function.DayOfYear;
import com.meidusa.amoeba.sqljep.function.Decode;
import com.meidusa.amoeba.sqljep.function.Floor;
import com.meidusa.amoeba.sqljep.function.Hash;
import com.meidusa.amoeba.sqljep.function.Hour;
import com.meidusa.amoeba.sqljep.function.IndistinctMatching;
import com.meidusa.amoeba.sqljep.function.Initcap;
import com.meidusa.amoeba.sqljep.function.Instr;
import com.meidusa.amoeba.sqljep.function.LastDay;
import com.meidusa.amoeba.sqljep.function.Length;
import com.meidusa.amoeba.sqljep.function.Lower;
import com.meidusa.amoeba.sqljep.function.Lpad;
import com.meidusa.amoeba.sqljep.function.Ltrim;
import com.meidusa.amoeba.sqljep.function.MakeDate;
import com.meidusa.amoeba.sqljep.function.MakeTime;
import com.meidusa.amoeba.sqljep.function.Microsecond;
import com.meidusa.amoeba.sqljep.function.Minute;
import com.meidusa.amoeba.sqljep.function.Modulus;
import com.meidusa.amoeba.sqljep.function.Month;
import com.meidusa.amoeba.sqljep.function.MonthName;
import com.meidusa.amoeba.sqljep.function.MonthsBetween;
import com.meidusa.amoeba.sqljep.function.NextDay;
import com.meidusa.amoeba.sqljep.function.Nvl;
import com.meidusa.amoeba.sqljep.function.PostfixCommand;
import com.meidusa.amoeba.sqljep.function.Power;
import com.meidusa.amoeba.sqljep.function.Replace;
import com.meidusa.amoeba.sqljep.function.Round;
import com.meidusa.amoeba.sqljep.function.Rpad;
import com.meidusa.amoeba.sqljep.function.Rtrim;
import com.meidusa.amoeba.sqljep.function.Second;
import com.meidusa.amoeba.sqljep.function.Sign;
import com.meidusa.amoeba.sqljep.function.SubDate;
import com.meidusa.amoeba.sqljep.function.SubTime;
import com.meidusa.amoeba.sqljep.function.Substring;
import com.meidusa.amoeba.sqljep.function.ToChar;
import com.meidusa.amoeba.sqljep.function.ToDate;
import com.meidusa.amoeba.sqljep.function.ToNumber;
import com.meidusa.amoeba.sqljep.function.Translate;
import com.meidusa.amoeba.sqljep.function.Trim;
import com.meidusa.amoeba.sqljep.function.Trunc;
import com.meidusa.amoeba.sqljep.function.Upper;
import com.meidusa.amoeba.sqljep.function.WeekOfYear;
import com.meidusa.amoeba.sqljep.function.Year;
import com.meidusa.amoeba.sqljep.variable.Variable;
import com.meidusa.amoeba.config.BeanObjectEntityConfig;
import com.meidusa.amoeba.config.ConfigurationException;
import com.meidusa.amoeba.config.DocumentUtil;
import com.meidusa.amoeba.context.ProxyRuntimeContext;
import com.meidusa.amoeba.net.DatabaseConnection;
import com.meidusa.amoeba.net.poolable.ObjectPool;
import com.meidusa.amoeba.parser.ParseException;
import com.meidusa.amoeba.parser.Parser;
import com.meidusa.amoeba.sqljep.RowJEP;
import com.meidusa.amoeba.util.Initialisable;
import com.meidusa.amoeba.util.InitialisationException;
import com.meidusa.amoeba.util.StringUtil;

/**
 * 
 * @author struct
 *
 */
public abstract class AbstractQueryRouter implements QueryRouter, Initialisable{

	private static Logger logger = Logger.getLogger(AbstractQueryRouter.class);
	final public static Map<String, PostfixCommand> ruleFunTab = new HashMap<String,PostfixCommand>();
	static{
		ruleFunTab.put("abs", new Abs());
	    ruleFunTab.put("power", new Power());
	    ruleFunTab.put("mod", new Modulus());
	    ruleFunTab.put("substr", new Substring());
	    ruleFunTab.put("sign", new Sign());
	    ruleFunTab.put("ceil", new Ceil());
	    ruleFunTab.put("floor", new Floor());
	    ruleFunTab.put("trunc", new Trunc());
	    ruleFunTab.put("round", new Round());
	    ruleFunTab.put("length", new Length());
	    ruleFunTab.put("concat", new Concat());
	    ruleFunTab.put("instr", new Instr());
	    ruleFunTab.put("trim", new Trim());
	    ruleFunTab.put("rtrim", new Rtrim());
	    ruleFunTab.put("ltrim", new Ltrim());
	    ruleFunTab.put("rpad", new Rpad());
	    ruleFunTab.put("lpad", new Lpad());
	    ruleFunTab.put("lower", new Lower());
	    ruleFunTab.put("upper", new Upper());
	    ruleFunTab.put("translate", new Translate());
	    ruleFunTab.put("replace", new Replace());
	    ruleFunTab.put("initcap", new Initcap());
	    ruleFunTab.put("value", new Nvl());
	    ruleFunTab.put("decode", new Decode());
	    ruleFunTab.put("to_char", new ToChar());
	    ruleFunTab.put("to_number", new ToNumber());
	    ruleFunTab.put("imatch", new IndistinctMatching());         // replacement for of Oracle's SOUNDEX
	    ruleFunTab.put("months_between", new MonthsBetween());
	    ruleFunTab.put("add_months", new AddMonths());
	    ruleFunTab.put("last_day", new LastDay());
	    ruleFunTab.put("next_day", new NextDay());
	    ruleFunTab.put("to_date", new ToDate());
	    ruleFunTab.put("case", new Case());                 // replacement for CASE WHEN digit = 0 THEN ...;WHEN digit = 1 THEN ...;ELSE ... END CASE
	    ruleFunTab.put("index", new Instr());                                               // maxdb
	    ruleFunTab.put("num", new ToNumber());                              // maxdb
	    ruleFunTab.put("chr", new ToChar());                                // maxdb
	    ruleFunTab.put("dayname", new DayName());                   // maxdb
	    ruleFunTab.put("adddate", new AddDate());                           // maxdb
	    ruleFunTab.put("subdate", new SubDate());                           // maxdb
	    ruleFunTab.put("addtime", new AddTime());                   // maxdb
	    ruleFunTab.put("subtime", new SubTime());                           // maxdb
	    ruleFunTab.put("year", new Year());                                         // maxdb
	    ruleFunTab.put("month", new Month());                                       // maxdb
	    ruleFunTab.put("day", new Day());                                                   // maxdb
	    ruleFunTab.put("dayofmonth", new Day());                            // maxdb
	    ruleFunTab.put("hour", new Hour());                                         // maxdb
	    ruleFunTab.put("minute", new Minute());                                     // maxdb
	    ruleFunTab.put("second", new Second());                             // maxdb
	    ruleFunTab.put("microsecond", new Microsecond());   // maxdb
	    ruleFunTab.put("datediff", new Datediff());                         // maxdb
	    ruleFunTab.put("dayofweek", new DayOfWeek());       // maxdb
	    ruleFunTab.put("weekofyear", new WeekOfYear());     // maxdb
	    ruleFunTab.put("dayofyear", new DayOfYear());               // maxdb
	    ruleFunTab.put("dayname", new DayName());                   // maxdb
	    ruleFunTab.put("monthname", new MonthName());       // maxdb
	    ruleFunTab.put("makedate", new MakeDate());         // maxdb
	    ruleFunTab.put("maketime", new MakeTime());         // maxdb
	    ruleFunTab.put("hash", new Hash());         //
	}
	
	/* 默认1000 */
	private LRUMap map;
	private Lock mapLock = new ReentrantLock(false);
	private int LRUMapSize = 10000;
	
	private Map<Table,TableRule> tableRuleMap = new HashMap<Table,TableRule>();
	private Map<String,Function> functionMap = new HashMap<String,Function>();
	private Map<String,PostfixCommand> ruleFunctionMap = new HashMap<String,PostfixCommand>();
	
	protected ObjectPool[] defaultPools;
	protected ObjectPool[] readPools;
	protected ObjectPool[] writePools;
	private String ruleConfig;
	private String functionConfig;
	private String ruleFunctionConfig;
	
	private String defaultPool;
	private String readPool;
	private String writePool;
	private boolean needParse = true;
	private boolean needEvaluate = true;
	
	public AbstractQueryRouter(){
		ruleFunctionMap.putAll(ruleFunTab);
	}
	
	public String getRuleConfig() {
		return ruleConfig;
	}
	
	/**
	 * 创建一个新的sql parser
	 * @param connection 
	 * @param sql
	 * @return Parser
	 */
	public  abstract Parser newParser(String sql);
	
	public void setRuleConfig(String ruleConfig) {
		this.ruleConfig = ruleConfig;
	}
	
	public void setReadPool(String readPool) {
		this.readPool = readPool;
	}

	public String getReadPool() {
		return readPool;
	}

	public String getWritePool() {
		return writePool;
	}

	public void setWritePool(String writePool) {
		this.writePool = writePool;
	}
	
	public ObjectPool[] doRoute(DatabaseConnection connection,String sql,boolean ispreparedStatment, Object[] parameters) {
		if(sql == null) return defaultPools;
		if(needParse){
			return selectPool(connection, sql,ispreparedStatment,parameters);
		}else{
			return this.defaultPools;
		}
		
	}
	
	/**
	 * 返回Query 被route到目标地址 ObjectPool集合
	 * 如果返回null，则是属于DatabaseConnection 自身属性设置的请求。
	 * @param connection
	 * @param sql
	 * @param parameters
	 * @return
	 */
	@SuppressWarnings("unchecked")
	protected ObjectPool[] selectPool(DatabaseConnection connection,String sql,boolean ispreparedStatment,Object[] parameters){
		List<String> poolNames = new ArrayList<String>();
		
		Statment statment = parseSql(connection,sql);

		if(statment != null){
			if(logger.isDebugEnabled()){
				Expression expression = statment.getExpression();
				logger.debug("Sql:["+sql +"] Expression=["+expression+"]");
			}
		}
		
		Map<Table,Map<Column,Comparative>> tables = null;
		DMLStatment dmlStatment = null;
		if(statment instanceof DMLStatment){
			dmlStatment = ((DMLStatment)statment);
			if(needEvaluate){
				tables = dmlStatment.evaluate(parameters);
			}
		}else if(statment instanceof PropertyStatment){
			setProperty(connection,(PropertyStatment)statment,parameters);
			return null;
		}else if(statment instanceof StartTansactionStatment){
			return null;
		}else if(statment instanceof CommitStatment){
			return null;
		}else if(statment instanceof RollbackStatment){
			return null;
		}
		
		if(tables != null && tables.size() >0){
			Set<Map.Entry<Table,Map<Column,Comparative>>> entrySet  = tables.entrySet();
			for(Map.Entry<Table,Map<Column,Comparative>> entry : entrySet){
				Map<Column,Comparative> columnMap = entry.getValue();
				TableRule tableRule = this.tableRuleMap.get(entry.getKey());
				
				/**
				 * 如果存在table Rule 则需要看是否有Rule
				 */
				if(tableRule != null){
					
					if(columnMap == null || ispreparedStatment){
						String[] pools = dmlStatment.isReadStatment()?tableRule.readPools:tableRule.writePools;
						if(pools == null){
							pools = tableRule.defaultPools;
						}
						for(String poolName : pools){
							if(!poolNames.contains(poolName)){
								poolNames.add(poolName);
							}