CDO框架架构

架构：

1、监听器WebApplicationListener 启动时，读取servicebus.xml内容，转化为String型

 

public void contextInitialized(ServletContextEvent arg0)
	{
		BusinessService app = BusinessService.getInstance();
		Return	ret = app.start();
		...
	}

 

public class BusinessService
{
	//静态对象
	private static BusinessService instance=new BusinessService();

	private BusinessService()
	{		
		serviceBus=new ServiceBus();
	}
	
	public static BusinessService getInstance()
	{//使用单列
		return instance;
	}

	//内部对象
	private ServiceBus serviceBus;	
	private boolean bIsRunning;
	
	public IServiceBus getServiceBus()
	{
		return this.serviceBus;
	}	
	
	public boolean isRunning()
	{
		return this.bIsRunning;
	}

	public Return start()
	{
		return start("servicebus.xml","UTF-8");
	}
	public Return start(String strServiceBusXMLFile,String encoding)
	{
		if(this.bIsRunning==true)
		{
			return Return.OK;
		}

		//将servicebus.xml内容解析为字符串
		String strBusinessServiceBusXML=Utility.readTextResource(strServiceBusXMLFile,encoding);
		Return ret=serviceBus.init(strBusinessServiceBusXML);
		if(ret.getCode()!=0)
		{
			return ret;
		}
		
		ret=serviceBus.start();
		if(ret.getCode()!=0)
		{
			serviceBus.destroy();
			return ret;
		}
		
		this.bIsRunning=true;

		return Return.OK;
	}
	public void stop()
	{
		if(this.bIsRunning==false)
		{
			return;
		}
		serviceBus.stop();
		serviceBus.destroy();
		
		this.bIsRunning=false;
	}

	public Return handleTrans(CDO cdoRequest,CDO cdoResponse)
	{
		Return ret=serviceBus.handleTrans(cdoRequest,cdoResponse);
		if(ret==null)
		{
			return Return.valueOf(-1,"Invalid request","System.Error");
		}
		
		return ret;
	}
	
	
}

 

 

start("servicebus.xml","UTF-8");

String strBusinessServiceBusXML=Utility.readTextResource(strServiceBusXMLFile,encoding);

InputStreamInputStreamReaderBufferReader  readLine

 

Return ret=serviceBus.init(strBusinessServiceBusXML);

 

 

2、serviceBus.init(strBusinessServiceBusXML) 方法

 

a、先解析XML内容字符串并转换为ServiceBus 对象

 

//使用castor将XML数据转换为业务对象  	
	
	public static ServiceBus fromXML(String strXML)  {
		StringReader reader = null;
		try	{
			  reader = new StringReader(strXML);
			  ServiceBus serviceBus = unmarshal(reader);
			  return serviceBus;
		}
		finally{ }	
	}

	public static ServiceBus unmarshal(Reader reader) {
		return (ServiceBus)Unmarshaller.unmarshal(ServiceBus.class, reader);
	}

 

 

b、从ServiceBus 对象中获取各配置项，同时解析赋值到对应的Map对象或其他对象中

 

this.strDefaultDataGroupId = serviceBus.getDataGroupId();
	 
	this.hmDataGroup = new HashMap(20);
	DataGroup[] dgs = serviceBus.getDataGroup();
	try
	{
	  for (int i = 0; i < dgs.length; i++)
	  {
		  /* DataGroup中init()方法
		  * 每一个DataGroup中有多个Database，每一个Database中有一个IDataEngine，
		  * 在IDataEngine中填充各数据库连接属性并打开连接
		  */
		CycleList clDataEngine = dgs[i].init();
		this.hmDataGroup.put(dgs[i].getId(), clDataEngine);
	  }
	}
	catch (Exception e)
	{
	  this.hmDataGroup.clear();
	  this.hmDataGroup = null;
	  this.logger.error("When parse DataGroup , caught exception: ", e);
	  return Return.valueOf(-1, "Init ServiceBus Failed: " + e.getLocalizedMessage());
	}

	
	//分别解析Plugin.xml文件
	int nPluginCount = serviceBus.getPluginXMLResourceCount();
    this.plugins = new ServicePlugin[nPluginCount];
    try
    {
      for (int i = 0; i < nPluginCount; i++)
      {
		//<PluginXMLResource>com/service/mobile/Plugin.xml</PluginXMLResource>标签中间的内容，即文件名
        String strXMLResource = serviceBus.getPluginXMLResource(i);
        if (this.logger.isInfoEnabled()) this.logger.info("loading " + strXMLResource + "..............................");
		
		//将各Plugin.xml文件内容解析为字符串
        String strXML = Utility.readTextResource(strXMLResource, "utf-8");
        if (strXML == null)
        {
          throw new Exception("Resource " + strXMLResource + " invalid");
        }
		
		//解析Plugin.xml内容为对应的ServicePlugin对象(同上方的解析为ServiceBus对象)
        com.cdoframework.cdolib.servicebus.schema.ServicePlugin servicePluginDefine = com.cdoframework.cdolib.servicebus.schema.ServicePlugin.fromXML(strXML);
        this.plugins[i] = new ServicePlugin();

        this.plugins[i].setServiceBus(this);
        this.plugins[i].setPublicDataGroup(this.hmDataGroup);
        this.plugins[i].setPublicNoSQLDataEngine(this.hmNoSQLDataEngine);
        this.plugins[i].init(i, this, servicePluginDefine);
      }
    }
    catch (Exception e)
    {
      this.plugins = null;

      this.logger.error("when parse plugin ,caught Exception: ", e);
      return Return.valueOf(-1, "Init ServiceBus Failed: " + e.getLocalizedMessage());
    }

 

 

 

各文件摘要：

 

Plugin.xml：

 

<?xml version="1.0" encoding="UTF-8"?>
<ServicePlugin xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="../../../../../../../Design/ServiceBus.xsd">
	<ServiceConfig Id="ActivityService"/>
	<DataService Id="ActivityService" Resource="com/service/activity/ActivitySQLTrans.xml"/>
	<TransService Id="ActivityService"  ClassPath="com.service.activity.ActivityService" />
		
	<ServiceConfig Id="WxActivityService"/>
	<DataService Id="WxActivityService" Resource="com/service/activity/WxActivitySQLTrans.xml"/>
	<TransService Id="WxActivityService"  ClassPath="com.service.activity.WxActivityService" />
	...	
</ServicePlugin>

 

 

servicebus.xml内容摘要：

 

<?xml version="1.0" encoding="utf-8"?>
<ServiceBus xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="../../Design/ServiceBus.xsd"
		DataGroupId="writeMall" NoSQLDBId="mainId">
		
		<!-- 多个dataGroup-->
		<DataGroup Weight="0" Num="1" Id="write0" ClassPath="com.lepus.businessCenter.DataEngine" Driver="com.mysql.jdbc.Driver" Charset="gbk">
			<Database URI="jdbc:mysql://192.168.3.230:3306/1more" LoadLevel="100">
				<User UserName="root" Password="123456"/>
				<ConnectionPool>
					<MaxConnectionsPerPartition>2</MaxConnectionsPerPartition>
					<MinConnectionsPerPartition>2</MinConnectionsPerPartition>
					<AcquireIncrement>2</AcquireIncrement>
					<ReleaseHelperThreads>2</ReleaseHelperThreads>
					<ConnectionTimeout>2000</ConnectionTimeout>
					<PartitionCount>2</PartitionCount>
				</ConnectionPool>
			</Database>
		</DataGroup>
		
		<!-- 多个PluginXMLResource-->		
		<PluginXMLResource>com/service/product/Plugin.xml</PluginXMLResource>
		<PluginXMLResource>com/service/platform/Plugin.xml</PluginXMLResource>
	
		<!-- 系统服务器,缓存，filter等等-->
		<EventProcessor MaxIdelTreadCount="3" MaxThreadCount="50" MaxWaitEventCount="1000" MinThreadCount="3"></EventProcessor>
		
		...
</ServiceBus>

 

 

 

ServiceBus.java:

 

//有多个的为ArrayList对象(_dataGroupList等)，单个的为字符串(_noSQLDBId)或其他对象(EventProcessor)

	package com.cdoframework.cdolib.servicebus.schema;

	import java.io.IOException;
	import java.io.Reader;
	import java.io.Serializable;
	import java.io.StringReader;
	import java.io.Writer;
	import java.util.ArrayList;
	import java.util.Enumeration;
	import org.apache.log4j.Logger;
	import org.exolab.castor.util.IteratorEnumeration;
	import org.exolab.castor.xml.MarshalException;
	import org.exolab.castor.xml.Marshaller;
	import org.exolab.castor.xml.Unmarshaller;
	import org.exolab.castor.xml.ValidationException;
	import org.exolab.castor.xml.Validator;
	import org.xml.sax.ContentHandler;

	public class ServiceBus  implements Serializable
	{
	  private String _dataGroupId;
	  private String _noSQLDBId;
	  private ArrayList _parameterList;
	  private ArrayList _dataGroupList;
	  private ArrayList _noSQLDBList;
	  private ClusterController _clusterController;
	  private EventProcessor _eventProcessor;
	  private ArrayList _pluginXMLResourceList;
	  
	  public ServiceBus()
	  {
		this._parameterList = new ArrayList();
		this._dataGroupList = new ArrayList();
		this._noSQLDBList = new ArrayList();
		this._pluginXMLResourceList = new ArrayList();
	  }

	  public void addDataGroup(DataGroup vDataGroup)
			throws IndexOutOfBoundsException
	  {
		this._dataGroupList.add(vDataGroup);
	  }

	  public void addDataGroup(int index, DataGroup vDataGroup)
			throws IndexOutOfBoundsException
	  {
		this._dataGroupList.add(index, vDataGroup);
	  }

	  public void clearDataGroup()
	  {
		this._dataGroupList.clear();
	  }

	   public boolean removeDataGroup(DataGroup vDataGroup) {
			boolean removed = this._dataGroupList.remove(vDataGroup);
			return removed;
	  }

	  public void setEventProcessor(EventProcessor eventProcessor)
	  {
		this._eventProcessor = eventProcessor;
	  }
	  
	  public EventProcessor getEventProcessor()
	  {
		return this._eventProcessor;
	  }
	  
	  public String[] getPluginXMLResource()
	  {
		int size = this._pluginXMLResourceList.size();
		String[] mArray = new String[size];
		for (int index = 0; index < size; index++) {
		  mArray[index] = ((String)this._pluginXMLResourceList.get(index));
		}
		return mArray;
	  }
	  
	  public static ServiceBus unmarshal(Reader reader)
			throws MarshalException, ValidationException
	  {
		return (ServiceBus)Unmarshaller.unmarshal(ServiceBus.class, reader);
	  }
	  ...
	}

 

 

DataGroup.java:

 

public class DataGroup  implements Serializable
	{
	  private String _id;
	  private String _num;
	  private String _weight;
	  private String _driver;
	  private String _charset;
	  private String _classPath;
	  private ArrayList _databaseList;

	  public DataGroup()
	  {
		this._databaseList = new ArrayList();
	  }

	  public void addDatabase(Database vDatabase)
		throws IndexOutOfBoundsException  {
		this._databaseList.add(vDatabase);
	  }

	  public void addDatabase(int index, Database vDatabase)
		throws IndexOutOfBoundsException
	  {
		this._databaseList.add(index, vDatabase);
	  }

	  public Database getDatabase(int index)
		throws IndexOutOfBoundsException
	  {
		if ((index < 0) || (index >= this._databaseList.size())) {
		  throw new IndexOutOfBoundsException();
		}

		return (Database)this._databaseList.get(index);
	  }

	  public Database[] getDatabase()
	  {
		int size = this._databaseList.size();
		Database[] mArray = new Database[size];
		for (int index = 0; index < size; index++) {
		  mArray[index] = ((Database)this._databaseList.get(index));
		}
		return mArray;
	  }
	  
	   public static DataGroup unmarshal(Reader reader)
			throws MarshalException, ValidationException
	  {
		return (DataGroup)Unmarshaller.unmarshal(DataGroup.class, reader);
	  }

	  /*
	  * DataGroup中init()方法
	  * 每一个DataGroup中有多个Database，每一个Database中有一个IDataEngine，
	  * 在IDataEngine中填充各数据库连接属性并打开连接
	  */
	  public CycleList<IDataEngine> init()
		throws Exception
	  {
		Database[] dbs = getDatabase();

		CycleList clDataEngine = new CycleList();
		for (int i = 0; i < dbs.length; i++)
		{
		  IDataEngine dataEngine = (IDataEngine)Class.forName(getClassPath()).newInstance();
		  dataEngine.setNum(getNum());
		  dataEngine.setWeight(getWeight());
		  dataEngine.setDriver(getDriver());
		  dataEngine.setURI(dbs[i].getURI());
		  dataEngine.setCharset(getCharset());
		  dataEngine.setLoadLevel(dbs[i].getLoadLevel());
		  dataEngine.setUserName(dbs[i].getUser().getUserName());
		  dataEngine.setPassword(dbs[i].getUser().getPassword());

		  Properties properties = null;
		  int nPropertyCount = dbs[i].getPropertyCount();
		  if (nPropertyCount > 0)
		  {
			properties = new Properties();
			for (int j = 0; j < nPropertyCount; j++)
			{
			  Property proper = dbs[i].getProperty(j);
			  properties.setProperty(proper.getName(), proper.getValue());
			}
		  }
		  if (properties != null)
		  {
			dataEngine.setProperties(properties);
		  }

		  ConnectionPool connPool = dbs[i].getConnectionPool();
		  if (connPool != null)
		  {
			dataEngine.setAcquireIncrement(connPool.getAcquireIncrement());
			dataEngine.setMaxConnectionsPerPartition(connPool.getMaxConnectionsPerPartition());
			dataEngine.setMinConnectionsPerPartition(connPool.getMinConnectionsPerPartition());
			dataEngine.setPartitionCount(connPool.getPartitionCount());
			dataEngine.setReleaseHelperThreads(connPool.getReleaseHelperThreads());
			dataEngine.setConnectionTimeout(connPool.getConnectionTimeout());
		  }

		  Return ret = dataEngine.open();
		  if (ret.getCode() != 0)
		  {
			throw new Exception("Could not create JDBC connection " + dataEngine.getURI());
		  }

		  clDataEngine.add(dataEngine);
		}

		return clDataEngine;
	  }
	  ...
	}

 

 

Database.java:

public class Database  implements Serializable
	{
	  private String _URI;
	  private int _loadLevel;
	  private boolean _has_loadLevel;
	  private User _user;
	  private ArrayList _propertyList;
	  private ConnectionPool _connectionPool;

	  public Property getProperty(int index)
			throws IndexOutOfBoundsException
	  {
		if ((index < 0) || (index >= this._propertyList.size())) {
		  throw new IndexOutOfBoundsException();
		}

		return (Property)this._propertyList.get(index);
	  }

	  public Property[] getProperty()
	  {
		int size = this._propertyList.size();
		Property[] mArray = new Property[size];
		for (int index = 0; index < size; index++) {
		  mArray[index] = ((Property)this._propertyList.get(index));
		}
		return mArray;
	  }  

	  public void setProperty(int index, Property vProperty)
		throws IndexOutOfBoundsException
	  {
		if ((index < 0) || (index >= this._propertyList.size())) {
		  throw new IndexOutOfBoundsException();
		}
		this._propertyList.set(index, vProperty);
	  }

	  public void setProperty(Property[] propertyArray)
	  {
		this._propertyList.clear();
		for (int i = 0; i < propertyArray.length; i++)
		  this._propertyList.add(propertyArray[i]);
	  }

	   public static Database unmarshal(Reader reader)
			throws MarshalException, ValidationException
	  {
		return (Database)Unmarshaller.unmarshal(Database.class, reader);
	  }
	  ...
	}

 

Property.java:

public class Property  implements Serializable
	{
	  private String _name;
	  private String _value;

	  public static Property unmarshal(Reader reader)
			throws MarshalException, ValidationException
	  {
		return (Property)Unmarshaller.unmarshal(Property.class, reader);
	  }	  
	  ...
	}

 

 

ConnectionPool.java:

public class ConnectionPool  implements Serializable
	{
	  private int _maxConnectionsPerPartition = 2;
	  private boolean _has_maxConnectionsPerPartition;
	  private int _minConnectionsPerPartition = 2;
	  private boolean _has_minConnectionsPerPartition;
	  private int _acquireIncrement = 2;
	  private boolean _has_acquireIncrement;
	  private int _releaseHelperThreads = 3;
	  private boolean _has_releaseHelperThreads;
	  private long _connectionTimeout = 1000L;
	  private boolean _has_connectionTimeout;
	  private int _partitionCount = 2;
	  private boolean _has_partitionCount;

	  public void deleteAcquireIncrement()
	  {
		this._has_acquireIncrement = false;
	  }

	  public static ConnectionPool unmarshal(Reader reader)
			throws MarshalException, ValidationException
	  {
		return (ConnectionPool)Unmarshaller.unmarshal(ConnectionPool.class, reader);
	  }

	  ...
	}