使用地圖觸發功能處理相干事件

本文介紹如何通過使用映射觸發器來處理一致性事件。基本上，建議使用Oracle Coherence中的分布式數據管理來研究Oracle Coherence API的基本配置和實現。
映射觸發器是Oracle Coherence提供最高度定制的緩存管理系統的最重要功能之一。 MapTrigger代表一個功能代理，該代理允許針對基礎地圖進行驗證，拒絕或修改變異操作。此外，它們還可以防止無效事務，增強安全性，提供事件日志記錄和審核以及收集有關數據修改的統計信息。

例如，我們有與NamedCache一起使用的代碼，并且我們想在將條目插入地圖之前更改條目的行為或內容。通過啟用映射觸發器，可以在不修改所有現有代碼的情況下進行此更改。

有兩種方法可以將“地圖觸發器”功能添加到應用程序：

1）一種MapTriggerListener可以使用具有指定的高速緩存來注冊MapTrigger
2）可以在coherence-cache-config.xml配置文件中使用類工廠機制

在以下示例應用程序中，通過遵循第一種方法來實現MapTrigger功能。創建了一個稱為OTV的新集群，并通過在該集群的兩個成員之間使用的用戶映射NamedCache對象分發了用戶bean。

二手技術：

JDK 1.6.0_35
Spring3.1.2
連貫性3.7.1 Maven的3.0.2 步驟1：建立已完成的專案

創建一個Maven項目，如下所示。（可以使用Maven或IDE插件來創建它）。

第2步：相干套餐

通過Coherence軟件包下載Coherence

步驟3：圖書館

首先，將Spring依賴項添加到Maven的pom.xml中。

<!-- Spring 3.1.2 dependencies --><dependency><groupId>org.springframework</groupId><artifactId>spring-core</artifactId><version>${spring.version}</version></dependency><dependency><groupId>org.springframework</groupId><artifactId>spring-context</artifactId><version>${spring.version}</version></dependency>

Coherence庫是手動安裝到Local Maven Repository的，其描述如下所示添加到pom.xml中。同樣，如果不使用Maven來管理項目，則可以將coherence.jar文件添加到classpath中。

<!-- Coherence library(from local repository) --><dependency><groupId>com.tangosol</groupId><artifactId>coherence</artifactId><version>3.7.1</version></dependency>

為了創建runnable-jar ，可以使用以下Maven插件。

<plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-shade-plugin</artifactId><version>1.3.1</version><executions><execution><phase>package</phase><goals><goal>shade</goal></goals><configuration><transformers><transformerimplementation='org.apache.maven.plugins.shade.resource.ManifestResourceTransformer'><mainClass>com.otv.exe.Application</mainClass></transformer><transformerimplementation='org.apache.maven.plugins.shade.resource.AppendingTransformer'><resource>META-INF/spring.handlers</resource></transformer><transformerimplementation='org.apache.maven.plugins.shade.resource.AppendingTransformer'><resource>META-INF/spring.schemas</resource></transformer></transformers></configuration></execution></executions></plugin>

步驟4：建立otv-coherence-cache-config.xml

第一個Coherence配置文件是otv-coherence-cache-config.xml 。它包含（分布式或復制的）緩存方案和緩存方案映射配置。創建的緩存配置應添加到coherence-cache-config.xml中 。

<?xml version='1.0'?><cache-config xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xmlns='http://xmlns.oracle.com/coherence/coherence-cache-config'xsi:schemaLocation='http://xmlns.oracle.com/coherence/coherence-cache-configcoherence-cache-config.xsd'><caching-scheme-mapping><cache-mapping><cache-name>user-map</cache-name><scheme-name>MapDistCache</scheme-name></cache-mapping></caching-scheme-mapping><caching-schemes><distributed-scheme><scheme-name>MapDistCache</scheme-name><service-name>MapDistCache</service-name><backing-map-scheme><local-scheme><unit-calculator>BINARY</unit-calculator></local-scheme></backing-map-scheme><autostart>true</autostart></distributed-scheme></caching-schemes></cache-config>

步驟5：創建tangosol-coherence-override.xml

第二個Coherence配置文件是tangosol-coherence-override.xml 。它包含集群，成員身份和可配置緩存工廠配置。

集群的第一個成員的tangosol-coherence-override.xml：

<?xml version='1.0'?><coherence xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xmlns='http://xmlns.oracle.com/coherence/coherence-operational-config'xsi:schemaLocation='http://xmlns.oracle.com/coherence/coherence-operational-config coherence-operational-config.xsd'><cluster-config><member-identity><cluster-name>OTV</cluster-name><role-name>OTV1</role-name></member-identity><unicast-listener><well-known-addresses><socket-address id='1'><address>x.x.x.x</address><port>8089</port></socket-address><socket-address id='2'><address>x.x.x.x</address><port>8090</port></socket-address></well-known-addresses><machine-id>1001</machine-id><address>x.x.x.x</address><port>8089</port><port-auto-adjust>true</port-auto-adjust></unicast-listener></cluster-config><configurable-cache-factory-config><init-params><init-param><param-type>java.lang.String</param-type><param-value system-property='tangosol.coherence.cacheconfig'>otv-coherence-cache-config.xml</param-value></init-param></init-params></configurable-cache-factory-config></coherence>

集群的第二個成員的tangosol-coherence-override.xml：

<?xml version='1.0'?><coherence xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xmlns='http://xmlns.oracle.com/coherence/coherence-operational-config'xsi:schemaLocation='http://xmlns.oracle.com/coherence/coherence-operational-config coherence-operational-config.xsd'><cluster-config><member-identity><cluster-name>OTV</cluster-name><role-name>OTV2</role-name></member-identity><unicast-listener>      <well-known-addresses><socket-address id='1'><address>x.x.x.x</address><port>8090</port></socket-address><socket-address id='2'><address>x.x.x.x</address><port>8089</port></socket-address></well-known-addresses><machine-id>1002</machine-id><address>x.x.x.x</address><port>8090</port><port-auto-adjust>true</port-auto-adjust></unicast-listener></cluster-config><configurable-cache-factory-config><init-params><init-param><param-type>java.lang.String</param-type><param-value system-property='tangosol.coherence.cacheconfig'>otv-coherence-cache-config.xml</param-value></init-param></init-params></configurable-cache-factory-config></coherence>

步驟6：創建applicationContext.xml

Spring配置文件applicationContext.xml已創建。

<beans xmlns='http://www.springframework.org/schema/beans'xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance'xsi:schemaLocation='http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans-3.0.xsd'><!-- Beans Declaration --><bean id='userCacheService' class='com.otv.srv.UserCacheService'></bean><bean id='userCacheUpdater' class='com.otv.exe.UserCacheUpdater'><property name='userCacheService' ref='userCacheService' /></bean></beans>

步驟7：創建用戶分類

創建了一個新的User Spring bean。該bean將分布在OTV集群中的兩個節點之間。對于序列化，已經實現了java.io.Serializable接口，但是可以實現PortableObject以獲得更好的性能。

package com.otv.user;import java.io.Serializable;/*** User Bean** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class User implements Serializable {private static final long serialVersionUID = -1963764656789800896L;private String id;private String name;private String surname;	public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public String getSurname() {return surname;}public void setSurname(String surname) {this.surname = surname;}@Overridepublic String toString() {StringBuilder strBuff = new StringBuilder();strBuff.append('id : ').append(id);strBuff.append(', name : ').append(name);strBuff.append(', surname : ').append(surname);return strBuff.toString();}
}

步驟8：建立IUserCacheService接口

為服務層創建了一個新的IUserCacheService接口，以公開緩存功能。

package com.otv.srv;import com.tangosol.net.NamedCache;/*** IUserCacheService Interface exposes User Cache operations** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public interface IUserCacheService {/*** Adds user entries to cache** @param Object key* @param Object value**/void addToUserCache(Object key, Object value);/*** Deletes user entries from cache** @param Object key**/void deleteFromUserCache(Object key);/*** Gets user cache** @retun NamedCache Coherence named cache*/NamedCache getUserCache();}

步驟9：建立UserCacheService IMPL類別

通過實現IUserCacheService創建UserCacheService 。

package com.otv.srv;import com.otv.listener.UserMapListener;
import com.otv.trigger.UserMapTrigger;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.NamedCache;
import com.tangosol.util.MapTriggerListener;/*** CacheService Class implements the ICacheService** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class UserCacheService implements IUserCacheService {private NamedCache userCache = null;private static final String USER_MAP = 'user-map';private static final long   LOCK_TIMEOUT = -1;public UserCacheService() {setUserCache(CacheFactory.getCache(USER_MAP));getUserCache().addMapListener(new UserMapListener());getUserCache().addMapListener(new MapTriggerListener(new UserMapTrigger()));}	/*** Adds user entries to cache** @param Object key* @param Object value**/public void addToUserCache(Object key, Object value) {// key is lockedgetUserCache().lock(key, LOCK_TIMEOUT);try {// application logicgetUserCache().put(key, value);} finally {// key is unlockedgetUserCache().unlock(key);}}/*** Deletes user entries from cache** @param Object key**/public void deleteFromUserCache(Object key) {// key is lockedgetUserCache().lock(key, LOCK_TIMEOUT);try {// application logicgetUserCache().remove(key);} finally {// key is unlockedgetUserCache().unlock(key);}}/*** Gets user cache** @retun NamedCache Coherence named cache*/public NamedCache getUserCache() {return userCache;}public void setUserCache(NamedCache userCache) {this.userCache = userCache;}}

步驟10：創建UserMapTrigger類

通過實現com.tangosol.util.MapTrigger接口，可以創建一個新的UserMapTrigger類。此觸發器在將條目插入到用戶映射中之前處理邏輯。

package com.otv.trigger;import org.apache.log4j.Logger;import com.otv.listener.UserMapListener;
import com.otv.user.User;
import com.tangosol.util.MapTrigger;/*** UserMapTrigger executes required logic before the operation is committed** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class UserMapTrigger implements MapTrigger {private static final long serialVersionUID = 5411263646665358790L;private static Logger logger = Logger.getLogger(UserMapListener.class);/*** Processes user cache entries** @param MapTrigger.Entry entry**/public void process(MapTrigger.Entry entry) {User user = (User) entry.getValue();String id = user.getId();String name = user.getName();String updatedName = name.toUpperCase();String surname = user.getSurname();String updatedSurname = surname.toUpperCase();if (!updatedName.equals(name)) {user.setName(updatedName);}if (!updatedSurname.equals(surname)) {user.setSurname(updatedSurname);}user.setId(user.getName() + '_' + user.getSurname());entry.setValue(user);logger.debug('UserMapTrigger processes the entry before committing. '+ 'oldId : ' + id+ ', newId : ' + ((User)entry.getValue()).getId()+ ', oldName : ' + name+ ', newName : ' + ((User)entry.getValue()).getName()+ ', oldSurname : ' + surname+ ', newSurname : ' + ((User)entry.getValue()).getSurname());}public boolean equals(Object o) {return o != null && o.getClass() == this.getClass();}public int hashCode() {return getClass().getName().hashCode();}
}

步驟11：建立USERMAPLISTENER IMPL類別

創建一個新的UserMapListener類。該偵聽器接收分布式用戶映射事件。

package com.otv.listener;import org.apache.log4j.Logger;import com.tangosol.util.MapEvent;
import com.tangosol.util.MapListener;/*** UserMapListener Class listens user cache events** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class UserMapListener implements MapListener {private static Logger logger = Logger.getLogger(UserMapListener.class);public void entryDeleted(MapEvent me) {logger.debug('Deleted Key = ' + me.getKey() + ', Value = ' + me.getOldValue());}public void entryInserted(MapEvent me) {logger.debug('Inserted Key = ' + me.getKey() + ', Value = ' + me.getNewValue());}public void entryUpdated(MapEvent me) {
//		logger.debug('Updated Key = ' + me.getKey() + ', New_Value = ' + me.getNewValue() + ', Old Value = ' + me.getOldValue());}
}

步驟12：創建CacheUpdater類

創建CacheUpdater類以添加新條目以緩存和監視緩存內容。

package com.otv.exe;import java.util.Collection;import org.apache.log4j.Logger;import com.otv.srv.IUserCacheService;
import com.otv.user.User;/*** CacheUpdater Class updates and prints user cache entries** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class UserCacheUpdater implements Runnable {private static Logger logger = Logger.getLogger(UserCacheUpdater.class);private IUserCacheService userCacheService;/*** Runs the UserCacheUpdater Thread**/public void run() {		//New User are created...User user = new User();//Only Name and Surname properties are set and Id property will be set at trigger level.user.setName('James');user.setSurname('Joyce');//Entries are added to cache...getUserCacheService().addToUserCache('user1', user);//		The following code block shows the entry which will be inserted via second member of the cluster
//      so it should be opened and above code block should be commented-out before the project is built.//		user.setName('Thomas');
//		user.setSurname('Moore');
//		getUserCacheService().addToUserCache('user2', user);//Cache Entries are being printed...printCacheEntries();}/*** Prints User Cache Entries**/@SuppressWarnings('unchecked')private void printCacheEntries() {Collection<User> userCollection = null;try {while(true) {userCollection = (Collection<User>)getUserCacheService().getUserCache().values();for(User user : userCollection) {logger.debug('Cache Content : '+user);}Thread.sleep(60000);}} catch (InterruptedException e) {logger.error('CacheUpdater is interrupted!', e);}}public IUserCacheService getUserCacheService() {return userCacheService;}public void setUserCacheService(IUserCacheService userCacheService) {this.userCacheService = userCacheService;}
}

步驟13：創建應用程序類

創建應用程序類以運行應用程序。

package com.otv.exe;import org.springframework.context.ApplicationContext;
import org.springframework.context.support.ClassPathXmlApplicationContext;/*** Application class starts the application** @author onlinetechvision.com* @since 29 Oct 2012* @version 1.0.0**/
public class Application {/*** Starts the application** @param  String[] args**/public static void main(String[] args) {ApplicationContext context = new ClassPathXmlApplicationContext('applicationContext.xml');UserCacheUpdater cacheUpdater = (UserCacheUpdater) context.getBean('userCacheUpdater');new Thread(cacheUpdater).start();}
}

nbsp;
步驟14：建立專案

生成OTV_Spring_Coherence_MapTrigger項目后，將創建OTV_Spring_Coherence_MapTrigger-0.0.1-SNAPSHOT.jar 。
重要說明：集群的成員具有不同的Coherence配置，因此應為每個成員分別構建項目。

步驟15：通過啟動集群成員來運行項目

在集群成員上運行OTV_Spring_Coherence-0.0.1-SNAPSHOT.jar文件后，將在第一個成員的控制臺上顯示以下輸出日志：

--A new cluster is created and First Member joins the cluster and adds a new entry to the cache.
29.10.2012 18:26:44 DEBUG (UserMapTrigger.java:49) - UserMapTrigger processes the entry before committing. oldId : null, newId : JAMES_JOYCE
, oldName : James, newName : JAMES, oldSurname : Joyce, newSurname : JOYCE
29.10.2012 18:26:44 DEBUG (UserMapListener.java:25) - Inserted Key = user1, Value = id : JAMES_JOYCE, name : JAMES, surname : JOYCE
29.10.2012 18:26:44 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : JAMES_JOYCE, name : JAMES, surname : JOYCE.......--Second Member joins the cluster and adds a new entry to the cache.
29.10.2012 18:27:33 DEBUG (UserMapTrigger.java:49) - UserMapTrigger processes the entry before committing. oldId : null, newId : THOMAS_MOORE,
oldName : Thomas, newName : THOMAS, oldSurname : Moore, newSurname : MOORE
29.10.2012 18:27:34 DEBUG (UserMapListener.java:25) - Inserted Key = user2, Value = id : THOMAS_MOORE, name : THOMAS, surname : MOORE.......--After second member adds a new entry, cache content is shown as below :
29.10.2012 18:27:44 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : THOMAS_MOORE, name : THOMAS, surname : MOORE
29.10.2012 18:27:45 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : JAMES_JOYCE, name : JAMES, surname : JOYCE
29.10.2012 18:28:45 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : THOMAS_MOORE, name : THOMAS, surname : MOORE
29.10.2012 18:28:45 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : JAMES_JOYCE, name : JAMES, surname : JOYCE

第二成員的控制臺：

--After Second Member joins the cluster and adds a new entry to the cache, cache content is shown as below and the members has got same entries :.
29.10.2012 18:27:34 DEBUG (UserMapListener.java:25) - Inserted Key = user2, Value = id : THOMAS_MOORE, name : THOMAS, surname : MOORE
29.10.2012 18:27:34 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : JAMES_JOYCE, name : JAMES, surname : JOYCE
29.10.2012 18:27:34 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : THOMAS_MOORE, name : THOMAS, surname : MOORE
29.10.2012 18:28:34 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : JAMES_JOYCE, name : JAMES, surname : JOYCE
29.10.2012 18:28:34 DEBUG (UserCacheUpdater.java:63) - Cache Content : id : THOMAS_MOORE, name : THOMAS, surname : MOORE

步驟16：下載

https://github.com/erenavsarogullari/OTV_Spring_Coherence_MapTrigger

參考： Online Technology Vision博客中的JCG合作伙伴 Eren Avsarogullari 使用地圖觸發功能進行的一致性事件處理。