proxyruntimecontext.java

mysql集群

/*
 * 	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.context;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

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

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.log4j.helpers.LogLog;
import org.w3c.dom.Attr;
import org.w3c.dom.Document;
import org.w3c.dom.Element;
import org.w3c.dom.NamedNodeMap;
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.net.poolable.MultipleLoadBalanceObjectPool;
import com.meidusa.amoeba.net.poolable.ObjectPool;
import com.meidusa.amoeba.net.poolable.PoolableObject;
import com.meidusa.amoeba.net.poolable.PoolableObjectFactory;
import com.meidusa.amoeba.config.BeanObjectEntityConfig;
import com.meidusa.amoeba.config.ConfigurationException;
import com.meidusa.amoeba.config.DBServerConfig;
import com.meidusa.amoeba.config.DocumentUtil;
import com.meidusa.amoeba.config.ParameterMapping;
import com.meidusa.amoeba.config.ProxyServerConfig;
import com.meidusa.amoeba.net.ConnectionManager;
import com.meidusa.amoeba.route.QueryRouter;
import com.meidusa.amoeba.util.Initialisable;
import com.meidusa.amoeba.util.InitialisationException;
import com.meidusa.amoeba.util.Reporter;
import com.meidusa.amoeba.util.StringUtil;

/**
 * 
 * @author <a href=mailto:[e-mail]>Struct chen</a>
 *
 */
public abstract class ProxyRuntimeContext implements Reporter{
	public static final String DEFAULT_SERVER_CONNECTION_MANAGER_CLASS = "com.meidusa.amoeba.net.AuthingableConnectionManager";
	public static final String DEFAULT_REAL_POOL_CLASS = "com.meidusa.amoeba.net.poolable.PoolableObjectPool";
	public static final String DEFAULT_VIRTUAL_POOL_CLASS = "com.meidusa.amoeba.server.MultipleServerPool";
	
	protected static Logger logger = Logger.getLogger(ProxyRuntimeContext.class);
	private static ProxyRuntimeContext context;
	public static ProxyRuntimeContext getInstance(){
		return context;
	}
	
	protected static void setInstance(ProxyRuntimeContext context){
		ProxyRuntimeContext.context = context;
	}
	
	protected abstract String getDefaultServerConnectionFactoryClassName();
	
	protected String getDefaultServerConnectionManagerClassName(){
		return DEFAULT_SERVER_CONNECTION_MANAGER_CLASS;
	}
	
	protected String getDefaultRealPoolClassName(){
		return DEFAULT_REAL_POOL_CLASS;
	}
	
	protected String getDefaultVirtualPoolClassName(){
		return DEFAULT_VIRTUAL_POOL_CLASS;
	}
	
	private Executor readExecutor;
	private Executor clientSideExecutor;
	private Executor serverSideExecutor;
	private Map<String,ConnectionManager> conMgrMap = new HashMap<String,ConnectionManager>();
	private Map<String,ObjectPool> poolMap = new HashMap<String,ObjectPool>();
	
	private Map<String,ConnectionManager> readOnlyConMgrMap = Collections.unmodifiableMap(conMgrMap);
	private Map<String,ObjectPool> readOnlyPoolMap = Collections.unmodifiableMap(poolMap);
	private ProxyServerConfig config ;
	
	private QueryRouter queryRouter;
	
	private String serverCharset;
	
	public String getServerCharset() {
		return serverCharset;
	}

	public void setServerCharset(String serverCharset) {
		this.serverCharset = serverCharset;
	}
	
	static class ReNameableThreadExecutor extends ThreadPoolExecutor{
		//Map<Thread,String> threadNameMap = new HashMap<Thread,String>();
		public ReNameableThreadExecutor(int poolSize) {
			super(poolSize, poolSize, Long.MAX_VALUE, TimeUnit.NANOSECONDS,new LinkedBlockingQueue<Runnable>());
		}
		
		/*protected void beforeExecute(Thread t, Runnable r) {
			if(r instanceof NameableRunner){
				NameableRunner nameableRunner = (NameableRunner)r;
				String name = t.getName();
				if(name != null){
					threadNameMap.put(t, t.getName());
					t.setName(nameableRunner.getRunnerName()+":"+t.getName());
				}
			}
		};
		protected void afterExecute(Runnable r, Throwable t) { 
			if(r instanceof NameableRunner){
				String name = threadNameMap.remove(Thread.currentThread());
				if(name != null){
					Thread.currentThread().setName(name);
				}
			}
		};*/
		
	}
	protected ProxyRuntimeContext(){
	}
	
	public Map<String,ConnectionManager> getConnectionManagerList(){
		return readOnlyConMgrMap;
	}
	
	public Executor getClientSideExecutor(){
		return clientSideExecutor;
	}
	
	public Executor getReadExecutor(){
		return readExecutor;
	}
	
	public Executor getServerSideExecutor(){
		return serverSideExecutor;
	}
	
	public Map<String,ObjectPool> getPoolMap(){
		return readOnlyPoolMap;
	}
	
	private List<Initialisable> initialisableList = new ArrayList<Initialisable>();
	public void init(String file){
		
		config = loadConfig(file);
		readExecutor = new ReNameableThreadExecutor(config.getReadThreadPoolSize());
		serverSideExecutor = new ReNameableThreadExecutor(config.getServerSideThreadPoolSize());
		clientSideExecutor = new ReNameableThreadExecutor(config.getClientSideThreadPoolSize());
		this.setServerCharset(config.getServerCharset());
		
		for(Map.Entry<String, BeanObjectEntityConfig> entry : config.getManagers().entrySet()){
			BeanObjectEntityConfig beanObjectEntityConfig = entry.getValue();
			try {
				ConnectionManager manager = (ConnectionManager) beanObjectEntityConfig.createBeanObject(false);
				manager.setName(entry.getKey());
				if(manager instanceof Initialisable){
					initialisableList.add((Initialisable)manager);
				}
				this.conMgrMap.put(entry.getKey(), manager);
			} catch (Exception e) {
				throw new ConfigurationException("manager instance error",e);
			}
		}
		
		for(Map.Entry<String, DBServerConfig> entry : config.getDbServers().entrySet()){
			DBServerConfig mysqlServerConfig = entry.getValue();
			try {
				BeanObjectEntityConfig poolConfig = mysqlServerConfig.getPoolConfig();
				ObjectPool pool = (ObjectPool)poolConfig.createBeanObject(false);
				if(pool instanceof Initialisable){
					initialisableList.add((Initialisable)pool);
				}
				if(mysqlServerConfig.getFactoryConfig() != null){
					PoolableObjectFactory factory = (PoolableObjectFactory)mysqlServerConfig.getFactoryConfig().createBeanObject(false);
					if(factory instanceof Initialisable){
						initialisableList.add((Initialisable)factory);
					}
					pool.setFactory(factory);
				}
				this.poolMap.put(entry.getKey(), pool);
			} catch (Exception e) {
				throw new ConfigurationException("manager instance error",e);
			}
		}
		
		if(config.getQueryRouterConfig() != null){
			BeanObjectEntityConfig queryRouterConfig = config.getQueryRouterConfig();
			try {
				queryRouter = (QueryRouter)queryRouterConfig.createBeanObject(false);
				if(queryRouter instanceof Initialisable){
					initialisableList.add((Initialisable)queryRouter);
				}
			} catch (Exception e) {
				throw new ConfigurationException("queryRouter instance error",e);
			}
		}
		initAllInitialisableBeans();
		initialisableList.clear();
		for(ConnectionManager conMgr :getConnectionManagerList().values()){
			conMgr.setExecutor(this.getReadExecutor());
			conMgr.start();
		}
		initPools();
	}
	
	
	protected void initPools(){
		for(Map.Entry<String, ObjectPool> entry: poolMap.entrySet()){
			ObjectPool pool = entry.getValue();
			if(pool instanceof MultipleLoadBalanceObjectPool){
				MultipleLoadBalanceObjectPool multiPool = (MultipleLoadBalanceObjectPool)pool;
				multiPool.initAllPools();
			}else{
				try{