使用 WebSphere eXtreme Scale 处理事件流

				
<objectGrids>
  <objectGrid name="StocksGrid">
    <bean id="ObjectGridEventListener" className="wxs.plugins.ObjectGridEventListener"/>
    <backingMap name="stockStreamMap" streamRef="stockStream"/>
    <backingMap name="forexStreamMap" streamRef="forexStream"/>            
    <backingMap name="stockViewMap" viewRef="stockView"/>            
    <backingMap name="forexViewMap" viewRef="forexView"/>

    <streamQuerySet name="stockMonitoringSQS">
        <stream name="stockStream" 
     		valueClass="wxs.streamquery.example.StockQuote" 
     		sql="create stream stockStream keyed by t (price DECIMAL (9,2), 
                  tickerSymbol VARCHAR(100));"
           	access="FIELD" >
        </stream>
        <stream name="forexStream" 
         	valueClass="wxs.streamquery.example.ForexQuote" 
          	sql="create stream forexStream keyed by t (rate DECIMAL (9,4), 
                    type VARCHAR(100) );"
               access="FIELD" >
        </stream>
        <view name="stockView"
          sql="CREATE VIEW stockView AS SELECT tickerSymbol, avg(price) as avgPrice FROM 
               (SELECT * FROM stockStream FETCH LATEST 5 MINUTES) group by tickerSymbol"
          access="FIELD">
        </view>
        <view name="forexView"                    
              sql="CREATE VIEW forexView AS SELECT type, avg(rate) as avgRate FROM 
                   SELECT * FROM forexStream FETCH LATEST 5 MINUTES) group by type"
              access="FIELD">
        </view>                 
    </streamQuerySet>
</objectGrid>
        
<objectGrid name="CustomerPortfolioGrid">
     <backingMap name="StockOwners" readOnly="false" lockStrategy="PESSIMISTIC"/>
     <backingMap name="Portfolios" readOnly="false" />      	
   <ObjectGrid>      	
</objectGrids>

				
<objectgridDeployment objectgridName="StocksGrid">
     <mapSet name="mapSet1" numberOfPartitions="1"
         minSyncReplicas="0" maxSyncReplicas="0"
         maxAsyncReplicas="0" numInitialContainers="1">
         <map ref="stockStreamMap"/>   
   	  <map ref="forexStreamMap"/>   
	  <map ref="forexViewMap"/>     
	  <map ref="stockViewMap"/>
     </mapSet>
</objectgridDeployment>

<objectgridDeployment objectgridName="CustomerPortfolioGrid">
     <mapSet name="mapSet1" numberOfPartitions="1"
         minSyncReplicas="0" maxSyncReplicas="0"
         maxAsyncReplicas="0" numInitialContainers="1">
         <map ref="StockOwners"/>   
         <map ref="Portfolios"/>
     </mapSet>
</objectgridDeployment>

				
static String[] stockList = {"AIG",  "AXP",  "ATT",  "INTC",  "MOT",  "IBM", "CSCO",
                             "JAVA", "MSFT", "ORCL", "HBOS", "HSBA",  "BAY", "BGY",
                             "BLND", "GSK", "ITV", "LAND",  "III", "BT"};

static float[] stockPrices = {49.12f, 52.40f, 24.95f, 23.56f,  10.07f, 123.29f, 26.92f,
                              13.14f, 29.43f, 21.55f, 4.71f, 8.82f, 2.09f, 7.14f,
                              8.04f, 11.53f, 0.61f, 14.67f, 8.97f, 2.33f}; 

static int[][] portfolioVolumes = {		
     	{1000, 500, 2000, 850, 4000, 2000, 7000, 2500, 5000, 9000, 2000},
{1000, 2500, 4000, 9000, 500, 900, 600, 2000},
	{8000, 900, 2500, 1000, 2000, 3000},
	{3000, 1000, 2500, 5500, 500, 2000, 7500, 1000},
	{3000, 5500, 4000, 500, 3000, 7000, 9000},
	{3500, 2500, 9000, 6500, 3000, 1000, 2500, 1000},
	{1000, 7000, 8000, 3000, 500, 3500, 5500},
	{9000, 3000, 5500, 1000, 8000, 4000, 3000, 2000, 1500},
	{2500, 500, 4000, 3000, 9000, 8000, 7000, 6000},
	{5500, 1000, 2500, 500, 3000, 8000, 500, 3000}};

static String[][] portfolioStocks = {	
{"AIG", "AXP", "ATT", "INTC", "MOT", "IBM", "CSCO", "JAVA", "MSFT", "HBOS"},
{"IBM", "CSCO", "JAVA", "MSFT", "ORCL", "BAY", "BGY", "BLND"},
{"IBM", "CSCO", "JAVA", "GSK", "ITV", "LAND"},
{"ATT", "INTC", "MOT", "IBM", "CSCO","III", "BT", "HBOS"},
{"AIG", "ATT", "INTC", "MOT", "JAVA", "ORCL","HSBA"},
{"ATT", "INTC", "MOT", "IBM", "CSCO",  "ORCL", "BAY", "BGY"},
{"AIG", "AXP", "INTC", "JAVA", "ORCL","BLND", "GSK"},
{"AIG", "AXP", "INTC", "IBM", "CSCO", "JAVA", "ORCL","ITV","LAND"},
{"AIG", "ATT", "MOT", "CSCO", "MSFT", "ORCL","III","BT"},
{"AXP", "ATT", "INTC", "IBM", "MSFT", "ORCL","GSK", "ITV"}};
   	
float initialDollarToPoundsForexRate = 1.9876f;

				
portfolioGrid = CustomerPortfolioGridHelper.getOG("CustomerPortfolioGrid");
stocksGrid = StocksGridHelper.getOG("StocksGrid");
    	    		
/* -------------------------------------------------------- */
/* Get ObjectGrid Session                                   */
/* -------------------------------------------------------- */
Session portfolioSession = portfolioGrid.getSession();

/* -------------------------------------------------------- */
/* Write Stock Ownership StockOwners Map                    */
/* -------------------------------------------------------- */
ObjectMap ownersMap = portfolioSession.getMap("StockOwners");	 
        
portfolioSession.begin();        
        
// Create array of StockOwners objects
for (int i = 0; i < stockList.length; i++)
{
	ArrayList customerIDs = new ArrayList();
	StockOwners stockOwners = new StockOwners(stockList[i]);
	stockOwners.setTickerSymbol(stockList[i]);
        	
	for (int j = 0; j < portfolioStocks.length; j++)
	{
		for (int k = 0; k < ScenarioInit.portfolioStocks[j].length; k++)
    		{
    			if (ScenarioInit.portfolioStocks[j][k].equals(stockList[i]))
    			{
    				customerIDs.add(String.valueOf(j + 1));
    			}
    		}
    	}
    	stockOwners.setCustomerIDs(customerIDs);
    	ownersMap.put(stockOwners.getTickerSymbol(), stockOwners);   
}
portfolioSession.commit();

				
synchronized static public ObjectGrid getOG(String objectGrid)
{
   if (og == null)
   {			
      try
      {			
	 URL url = Thread.currentThread().getContextClassLoader().
		getResource("SQ_ObjectGrid.xml");
								
	 ClientClusterContext ccc = bjectGridManagerFactory.getObjectGridManager().
		connect("localhost:6000", null,url);
								
	 ObjectGridManager ogm = ObjectGridManagerFactory.getObjectGridManager();
	 
         og = ogm.getObjectGrid(ccc,objectGrid);
			
      }
      catch (Exception e)
      {
	  e.printStackTrace();
      }			
   }
   else
   {
      System.out.println("StocksGridHelper.getOG() - OG already initialised");
   }		
   return og;
}

				
ObjectMap portfoliosMap = portfolioSession.getMap("Portfolios");
        
for (int j = 0; j < ScenarioInit.portfolioStocks.length; j++)
{
 	float portfolioValuation = 0.0f;
 	Portfolio portfolio = new Portfolio(String.valueOf(j + 1));
 	ArrayList<StockHolding> stockHoldingList = new ArrayList<StockHolding>();	
    		
 	portfolioSession.begin();     
    		
 	for (int k = 0; k < ScenarioInit.portfolioStocks[j].length; k++)
 	{
 	   int numUnits = ScenarioInit.portfolioVolumes[j][k];
 	   StockHolding stockHolding = 
           new StockHolding(ScenarioInit.portfolioStocks[j][k]);
    	   stockHolding.setNumUnits(numUnits);
    				
    	   if (isUSStock(ScenarioInit.portfolioStocks[j][k]))
    	   {
    	      stockValuation = 
                   numUnits * initialStockPrice(portfolioStocks[j][k])/
                                                  initialDollarToPoundsForexRate;
    	   }
    	   else
    	   {
    	      stockValuation = numUnits * initialStockPrice(portfolioStocks[j][k]);
    	   }
    	   stockHolding.setLatestValuation(stockValuation);
    	   stockHoldingList.add(stockHolding);
    	   portfolioValuation += stockValuation;    			
    	}
    	portfolio.setStocks(stockHoldingList);
    	portfolio.setInitialValuation(portfolioValuation);    			
    	portfoliosMap.insert(portfolio.getCustomerID(), portfolio);  
}
portfolioSession.commit();

				
Session stockStreamSession = stocksGrid.getSession();

ObjectMap stockStreamMap = stockStreamSession.getMap("stockStreamMap");
ObjectMap stockViewMap = stockStreamSession.getViewMap("stockViewMap");
    		
stockStreamSession.begin();
        
StockQuote.QuoteCurrency currency = null;
    		
for (int i = 0; i < stockList.length; i++)
{
 	if (i < 10)
 	{
 		currency = StockQuote.QuoteCurrency.USDOLLAR;
 	}
 	else
 	{
 		currency = StockQuote.QuoteCurrency.GBPOUND;
 	}

 	stockStreamMap.insert(stockList[i], new StockQuote(stockList[i], 
      stockPrices[i], 
      currency));  
}

    		
// Forex
forexStreamMap.insert("USDOLLAR_TO_GBPOUND", new ForexQuote("USDOLLAR_TO_GBPOUND",
        initialDollarToPoundsForexRate));

stockStreamSession.commit();

				
public void transactionEnd(String txid, boolean isWriteThruEnabled,
   boolean committed, Collection changes) 
{

 Iterator logSequenceIterator = changes.iterator();

 while (logSequenceIterator.hasNext())
 {
    LogSequence logSequence = (LogSequence)logSequenceIterator.next();
    String mapName = logSequence.getMapName();
    Iterator forexChangeIterator = logSequence.getAllChanges();

    if ((mapName.equals("forexViewMap") && logSequence.isDirty()))
    {

	while (changeIterator.hasNext())
	{
	  try
	  {
	   LogElement logElement = (LogElement)changeIterator.next();
	   Tuple forexKeyTuple = (Tuple)(logElement.getCacheEntry().getKey());
	   String key = (String)forexKeyTuple.getAttribute(0);
	   ogSession = instrumentGrid.getSession();
	   ObjectMap forexViewMap = ogSession.getViewMap("forexViewMap");
	   EntityMetadata emd = forexViewMap.getEntityMetadata();
	   TupleMetadata keyMD = emd.getKeyMetadata();
	   Tuple forexTuple = keyMD.createTuple();
	   forexTuple.setAttribute(0, "USDOLLAR_TO_GBPOUND"); //Could use key value
	   Tuple forexValueTuple = (Tuple)forexViewMap.get(forexTuple);
	   timeAveragedForexValuation = (Float)forexValueTuple.getAttribute(0);
	}
	catch ( Exception e)
      {......}
    }
  }
 }
}

				
else if (mapName.equals("stockViewMap") && logSequence.isDirty())
{
   while (changeIterator.hasNext())
   {
      try
      {
	   LogElement logElement = (LogElement)changeIterator.next();
	   LogElement.Type type = logElement.getType();	
	   Tuple keyTuple = (Tuple)logElement.getCacheEntry().getKey();
	   String key = (String)keyTuple.getAttribute(0);

	   Session instrumentSession = instrumentGrid.getSession();
	   instrumentSession.beginNoWriteThrough();
	   ObjectMap stockViewMap = instrumentSession.getViewMap("stockViewMap");

	   EntityMetadata emd = stockViewMap.getEntityMetadata();
	   TupleMetadata keyMD = emd.getKeyMetadata();
	   Tuple ibmKey = keyMD.createTuple();
	   ibmKey.setAttribute(0, key);

	   Tuple ibmValue = (Tuple) stockViewMap.get(ibmKey);
	   Float timeAveragedValuation = (Float)ibmValue.getAttribute(0);

 	   info("Stock Ticker Event :: Average price in the last 5 minutes for " +
                                               key + " = " + timeAveragedValuation);

	   instrumentSession.commit();

				
/* ------------------------------------------------------------- */
/* Retrieve List of Portfolios that hold this stock              */
/* ------------------------------------------------------------- */
ObjectMap portfoliosMap = portfolioSession.getMap("Portfolios");

StockOwners stockOwners = (StockOwners) holdingsMap.get(key);

ArrayList portfolioIDs = stockOwners.getCustomerIDs();

Iterator portfolioIterator = portfolioIDs.iterator();

				
while (portfolioIterator.hasNext())
{
portfolioSession.begin();
	String portfolioID = (String)portfolioIterator.next();	
	Portfolio portfolio = (Portfolio)portfoliosMap.get(portfolioID);
	ArrayList stocksList = portfolio.getStocks();
	ArrayList updatedStocksList = new ArrayList();
	Iterator stocksIterator = stocksList.iterator();

	float newPortfolioValuation = 0.0f;
	while (stocksIterator.hasNext())
	{
		StockHolding stockHolding = (StockHolding) stocksIterator.next();
		ibmKey.setAttribute(0, stockHolding.getTickerSymbol());
		ibmValue = (Tuple) stockViewMap.get(ibmKey);
		timeAveragedValuation = (Float)ibmValue.getAttribute(0);
		info("Stock Ticker Event :: Average price in the last 5 minutes for " 
			+ stockHolding.getTickerSymbol() + " = " + timeAveragedValuation);

		boolean usStockFlag = isUSStock(stockHolding.getTickerSymbol());
		if (usStockFlag)
	{
	   newStockValuation = stockHolding.getNumUnits() *
             timeAveragedValuation.floatValue()/timeAveragedForexValuation;
		}
		else
		{			        			
		   newStockValuation = stockHolding.getNumUnits() * 
                                                timeAveragedValuation.floatValue();
		}
		newPortfolioValuation += newStockValuation;
	}
	portfolioSession.commit();

				
float relativeChange = newPortfolioValuation/portfolio.getInitialValuation();

/* ---------------------------------------------------------------- */
/* If relative portfolio valuation  is > 1.10 or < 0.90 raise Alert */
/* ---------------------------------------------------------------- */
if ((relativeChange > 1.10) || (relativeChange < 0.90))
{
info("!************************************************************");
	info("!*");
	info("!* Portfolio Valuation Alert");
	info("!* Portfolio ID = " + portfolioID );
	info("!* Initial Valuation = " + portfolio.getInitialValuation());
	info("!* New Valuation = " + newPortfolioValuation);
	info("!* Relative Valuation Change = " + relativeChange);
	info("!*");
	info("!* Alert Triggering Stock = " + key);
	info("!*");
	info("!************************************************************");
}

				
ObjectMap stockMap = getObjectMap("stockStreamMap");

for (int i = 0; i < numEvents; i++)
{
int stockIndex = random(stockList.length);
	int priceIndex = random(stockPrices[stockIndex].length);
	float stockPrice = stockPrices[stockIndex][priceIndex];

	if (stockIndex < 10)
	{
		quoteCurrency = StockQuote.QuoteCurrency.USDOLLAR;
	}
	else
	{
		quoteCurrency = StockQuote.QuoteCurrency.GBPOUND;
	}

	StockQuote stockQuote = new StockQuote(stockList[stockIndex],
		stockPrice, quoteCurrency );

	stockMap.update(stockQuote.getTickerSymbol(), stockQuote);

	Thread.sleep(sleepTime);
}

				
call "%~dp0setupCmdLine.bat"

"C:\IBM\ObjectGrid\java\jre/bin/java" "-Xmx512m" 
"-classpath" "C:\IBM\ObjectGrid\java\lib\tools.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\objectgrid.jar;
C:\IBM\ObjectGrid\ObjectGrid\session\lib\sessionobjectgrid.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\cglib.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\ogstreamquery.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\castor.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\commons-io.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\mx4j.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\mx4j-remote.jar;
C:\IBM\ObjectGrid\ObjectGrid\lib\mx4j-tools.jar;
C:\IBM\ObjectGrid\ObjectGrid\properties;
c:\IBM\ObjectGrid\EventProcessing\EventStreamProcessing.jar" 

"com.ibm.ws.objectgrid.InitializationService" %1 
-objectgridFile c:\IBM\ObjectGrid\EventProcessing\ObjectGrid.xml 
-deploymentPolicyFile c:\IBM\ObjectGrid\EventProcessing\ObjectGridDeployment.xml 
-catalogServiceEndpoints localhost:6000

				
[22/05/08 11:31:47:671 BST]   fc00fc ServerImpl    I CWOBJ1001I: ObjectGrid Server c0 is
 ready to process requests.
[22/05/08 11:31:48:312 BST] 4b8e4b8e ReplicatedPar I CWOBJ1511I: CustomerPortfolioGrid:I
BM_SYSTEM_ENTITYMANAGER_MAPSET:0 (primary) is open for business.
[22/05/08 11:31:48:343 BST]  4d204d2 ReplicatedPar I CWOBJ1511I: CustomerPortfolioGrid:m
apSet1:0 (primary) is open for business.
[22/05/08 11:31:48:390 BST] 1d781d78 PeerManagerIm I CWOBJ8601I: PeerManager found peers
 of size 1
[22/05/08 11:31:48:390 BST] 1d781d78 ServerAgent   I CWOBJ7206I: New leader is (9.145.21
.53:3912). Old leader was (<null>).
[22/05/08 11:31:48:390 BST] 1d781d78 ServerAgent   I CWOBJ7203I: Leader changed.  New le
ader (9.145.21.53:3912) is elected in core group (DefaultZoneCG0) and re
ported to catalog server.
[22/05/08 11:31:48:515 BST]  4d204d2 ReplicatedPar I CWOBJ1511I: StocksGrid:mapSet1:0 (p
rimary) is open for business.
[22/05/08 11:31:48:578 BST] 4b8e4b8e ReplicatedPar I CWOBJ1511I: StocksGrid:IBM_SYSTEM_E
NTITYMANAGER_MAPSET:0 (primary) is open for business.

				
java "-classpath" "c:\IBM\ObjectGrid\ObjectGrid\lib\castor.jar;
c:\IBM\ObjectGrid\ObjectGrid\lib\commons-io.jar;
c:\IBM\ObjectGrid\ObjectGrid\lib\objectgrid.jar;
c:\IBM\ObjectGrid\ObjectGrid\lib\ogstreamquery.jar;
c:\IBM\ObjectGrid\java\lib\tools.jar;EventStreamProcessing.jar"
wxs.streamquery.example.ScenarioInit