Toplink分布式 Session Cache同步的方法oracle官方默認提供了JMS和RMI兩種實現方式，當然用戶也可以自定方法，自定義一個Transport Manager Class, 具體可參見：
http://download-west.oracle.com/docs/cd/B25221_04/web.1013/b13593/cachun003.htm
Oracle Coherence是用來實現Data Grid的framework。到目前為止，它還不能和Toplink的native版本整合，但是已經可以和Toplink Essential版本整合。Toplink升級版EclipseLink也承諾將會提供一個公共的接口讓用戶添加第三方的Cache Cluster到EclipseLink中來作為L2 Cache，但目前尚未實現。把coherence作為toplink session cache的transport manager是因為認為coherence的TCMP集群協議要比普通的jms和rmi的通信協議快。具體還有待進一步測試比較、分析。在目前我們無法使用coherence data grid作為toplink的L2 Cache情況下，又無法忍受使用JMS和RMI帶來的性能上的問題，使用一個自定義的Transport Manager是一個很好的嘗試。
Coherence介紹：
http://wiki.tangosol.com/display/COH/Oracle+Coherence+Knowledge+Base+Home
我們用Coherence實現Toplink Coordinated Cache，以下是這個demo的具體實現過程，它也僅僅是個demo。對于如何自定義toplink的cache coordinator還是很有幫助的。
1 Demo是用maven2+archifactory構建的，也使用了ant來作為部署時替換文本內容的工具。
2 Coherence配置
Coherence需要兩個配置文件： tangosol-coherence.xml和coherence-cache-config,并且用戶可以自己提供tangosol-coherence-override-dev.xml來重寫tangosol-coherence.xml的部分內容，根據不同的環境定義不同的tangosol-coherence.xml文件。可以參見tangosol的網站來進行配置。
coherence-cache-config,的配置：
<?xml version="1.0"?>
<!DOCTYPE cache-config SYSTEM "cache-config.dtd">
<cache-config>
<caching-scheme-mapping>
<cache-mapping>
<cache-name>toplinkSyn</cache-name>
<scheme-name>DistributedCacheScheme</scheme-name>
</cache-mapping>
</caching-scheme-mapping>
<caching-schemes>
<distributed-scheme>
<scheme-name>DistributedCacheScheme</scheme-name>
<service-name>DistributedCache</service-name>
<backing-map-scheme>
<local-scheme>
<scheme-ref>DistributedMap</scheme-ref>
</local-scheme>
</backing-map-scheme>
<backup-count>0</backup-count>
<autostart>true</autostart>
</distributed-scheme>
<local-scheme>
<scheme-name>DistributedMap</scheme-name>
<eviction-policy>LRU</eviction-policy>
<high-units>10000</high-units>
<expiry-delay>1D</expiry-delay>
<flush-delay>1D</flush-delay>
<cachestore-scheme></cachestore-scheme>
</local-scheme>
</caching-schemes>
</cache-config>
tangosol-coherence-override-dev.xml的配置
<?xml version='1.0'?>
<!--
This operational configuration override file is set up for use with Coherence in
a development mode.
-->
<coherence xml-override="/tangosol-coherence-override.xml">
<cluster-config>
<member-identity>
<cluster-name
system-property="tangosol.coherence.cluster">
Toplink Cache Synchronization
</cluster-name>
<member-name system-property="tangosol.coherence.member">
@{coherence.member.name}
</member-name>
<role-name>cache servers</role-name>
</member-identity>
<unicast-listener>
<well-known-addresses>
<socket-address id="1">
<address system-property="tangosol.coherence.wka">
146.222.51.20
</address>
<port
system-property="tangosol.coherence.wka.port">
8088
</port>
</socket-address>
<socket-address id="2">
<address system-property="tangosol.coherence.wka">
146.222.51.20
</address>
<port
system-property="tangosol.coherence.wka.port">
8089
</port>
</socket-address>
</well-known-addresses>
<address system-property="tangosol.coherence.localhost">
@{coherence.local.address}
</address>
<port system-property="tangosol.coherence.localport">
@{coherence.local.port}
</port>
</unicast-listener>
<authorized-hosts>
<host-address></host-address>
<host-range>
<from-address>146.222.51.0</from-address>
<to-address>146.222.51.255</to-address>
</host-range>
</authorized-hosts>
<packet-publisher>
<packet-delivery>
<timeout-milliseconds>30000</timeout-milliseconds>
</packet-delivery>
</packet-publisher>
</cluster-config>
<logging-config>
<destination>stderr</destination>
<severity-level
system-property="tangosol.coherence.log.level">
5
</severity-level>
<character-limit
system-property="tangosol.coherence.log.limit">
0
</character-limit>
</logging-config>
</coherence>
本demo會使用ant替換@{}中的內容。替換的key value值對需要在build.properties文件中給出。如：
coherence.member.name = tts-server2
coherence.local.address = 146.222.51.20
coherence.local.port = 8089
3 toplink配置
Demo使用了toplink tutorial的范例作為一個example來演示結果的正確性。下載地址：
http://www.oracle.com/technology/products/ias/toplink/doc/1013/main/_html/prt_tut.htm
4 demo的配置文件tts.properties：
#one of jms, rmi, coherence or set it blank
toplink.cache.type = coherence
#the name of toplink command channel
toplin.command.channel = OOCLToplinkCoherence
第一個參數用來根據不同的情況使用不同toplink session的配置文件。第二個參數是toplink Command Channel的名字，唯一標識一個toplink cluster。
3 主要代碼
1) CoherenceTransportManager繼承TransportManager實現自定義的Transport Manager Class
package com.oocl.isdc.sha.frm.tts.remotecommand;
import oracle.toplink.internal.remotecommand.RemoteConnection;
import oracle.toplink.remotecommand.DiscoveryManager;
import oracle.toplink.remotecommand.RemoteCommandManager;
import oracle.toplink.remotecommand.ServiceId;
import oracle.toplink.remotecommand.TransportManager;
import com.oocl.isdc.sha.frm.tts.cohererence.cache.CoherenceCache;
public class CoherenceTransportManager extends TransportManager 
{
protected CoherenceCache cache;
public CoherenceTransportManager(RemoteCommandManager rcm) {
this.rcm = rcm;
this.initialize();
}
public void initialize() {
super.initialize();
this.cache = new CoherenceCache();
}
/** *//**
* When get a session, toplink will call this method to listen to
* remote connection, and when some object is changed, it will get
* the chages.
*/
public void connectBackToRemote(RemoteConnection connection) {
CoherenceRemoteConnection coherenceConnection = (CoherenceRemoteConnection)connection;
coherenceConnection.becomeMapListener();
}
public void createLocalConnection() {
CoherenceRemoteConnection connection = new CoherenceRemoteConnection(rcm, cache);
addConnectionToExternalService(connection);
}
public RemoteConnection createConnection(ServiceId serviceId) {
return null;
}
public void removeLocalConnection() {
this.localConnection = null;
}
public DiscoveryManager createDiscoveryManager() {
return new CoherenceDiscoveryManager(rcm);
}
public CoherenceCache getCache() {
return cache;
}
public String getServiceUrl() {
return cache.getUrl();
}
}
2) CoherenceRemoteConnection繼承RemoteConnection從一個和Transport Mnager相關的連接。
1package com.oocl.isdc.sha.frm.tts.remotecommand;
2
3import oracle.toplink.exceptions.CommunicationException;
4import oracle.toplink.internal.remotecommand.RemoteConnection;
5import oracle.toplink.remotecommand.Command;
6import oracle.toplink.remotecommand.RemoteCommandManager;
7
8import com.oocl.isdc.sha.frm.tts.cohererence.cache.CoherenceCache;
9import com.tangosol.util.MapEvent;
10import com.tangosol.util.MapListener;
11
12public class CoherenceRemoteConnection extends RemoteConnection implements MapListener {
13
14 /** *//** Comment for <code>serialVersionUID</code> */
15 private static final long serialVersionUID = 8527315103990557963L;
16
17 private CoherenceCache cache;
18
19 private RemoteCommandManager rcm;
20
21 public CoherenceRemoteConnection(RemoteCommandManager rcm, CoherenceCache cache) {
22 this.serviceId = rcm.getServiceId();
23 this.rcm = rcm;
24 this.cache = cache;
25 }
26
27 /** *//**
28 * When some object in toplink session cache is chaged, it will callback this
29 * method to put the changed object infomation to coherence cache
30 */
31 public Object executeCommand(Command command) throws CommunicationException {
32 cache.putCache(command.getServiceId(), command);
33 return null;
34 }
35
36 @SuppressWarnings("unchecked")
37 protected void processObject(Object object) {
38 Command command = null;
39 if (object instanceof Command) {
40 command = (Command) object;
41 if (command.getServiceId().getChannel().equals(serviceId.getChannel())) {
42 rcm.processCommandFromRemoteConnection(command);
43 }
44 } else if (null == object) {
45
46 } else {
47
48 }
49
50 }
51
52 public void close() {
53 this.cache.removeMapListener(this);
54 }
55
56 public void becomeMapListener() {
57 this.cache.addMapListener(this);
58 }
59
60 public void entryDeleted(MapEvent arg0) {
61 }
62
63 /** *//**
64 * When an object is inserted into coherence, this method of
65 * listener will be called
66 */
67 public void entryInserted(MapEvent event) {
68 processObject(event.getNewValue());
69 }
70
71 /** *//**
72 * When an object in coherence cache is updated, this method
73 * of listener will be called
74 */
75 public void entryUpdated(MapEvent event) {
76 processObject(event.getNewValue());
77 }
78
79}
80
3) CoherenceSessionHelper將transport manager添加到當前session中支持server和database兩種session
1package com.oocl.isdc.sha.frm.tts.remotecommand;
2
3import oracle.toplink.remotecommand.CommandProcessor;
4import oracle.toplink.remotecommand.RemoteCommandManager;
5import oracle.toplink.sessions.DatabaseSession;
6import oracle.toplink.sessions.Session;
7import oracle.toplink.threetier.Server;
8
9import com.oocl.isdc.sha.frm.tts.config.TTSConfigConstants;
10import com.oocl.isdc.sha.frm.tts.config.TTSConfigParser;
11
12public class CoherenceSessionHelper {
13 public static Session addCoherenceTransportManagerToSession(Session session) {
14 RemoteCommandManager commandMgr = new RemoteCommandManager((CommandProcessor) session);
15 commandMgr.setChannel((String) TTSConfigParser.getInstance().get(
16 TTSConfigConstants.TOPLINK_COMMAND_CHANNEL_KEY));
17 CoherenceTransportManager tm = new CoherenceTransportManager(commandMgr);
18 tm.setInitialContextFactoryName(TTSConfigConstants.TTS_CONTEXT_FACTOYR_NAME);
19 commandMgr.setUrl(tm.getServiceUrl());
20 commandMgr.setTransportManager(tm);
21 tm.setShouldRemoveConnectionOnError(true);
22 if (session instanceof Server) {
23 ((Server) session).setCommandManager(commandMgr);
24 ((Server) session).setShouldPropagateChanges(true);
25 ((Server) session).getCommandManager().initialize();
26 } else if (session instanceof DatabaseSession) {
27 ((DatabaseSession) session).setCommandManager(commandMgr);
28 ((DatabaseSession) session).setShouldPropagateChanges(true);
29 ((DatabaseSession) session).getCommandManager().initialize();
30 } else {
31 throw new IllegalArgumentException("Session must be a server or database session");
32 }
33 return session;
34 }
35}
36
4) CoherenceCache:自定義一個Coherence NamedCache
package com.oocl.isdc.sha.frm.tts.cohererence.cache;
import java.net.InetAddress;
import java.util.Collection;
import com.oocl.isdc.sha.frm.tts.config.TTSConfigConstants;
import com.tangosol.net.CacheFactory;
import com.tangosol.net.Cluster;
import com.tangosol.net.Member;
import com.tangosol.net.NamedCache;
import com.tangosol.net.Service;
import com.tangosol.util.MapListener;
public class CoherenceCache {
private NamedCache namedCache;
public CoherenceCache() {
this.namedCache = CacheFactory.getCache("toplinkSyn");
}
public void putCache(Object key, Object value) {
namedCache.put(key, value);
}
public Object retrieveCache(Object key) {
return namedCache.get(key);
}
public NamedCache getNamedCache() {
return namedCache;
}
public void addMapListener(MapListener listener) {
this.namedCache.addMapListener(listener);
}
public void removeMapListener(MapListener listener) {
this.namedCache.removeMapListener(listener);
}
@SuppressWarnings("unchecked")
public Collection retrieveCacheAll() {
return this.namedCache.values();
}
public String getUrl() {
Member member = getLocalMember();
InetAddress address = member.getAddress();
String ipAddress = address.getHostAddress();
int port = member.getPort();
StringBuffer sb = new StringBuffer(TTSConfigConstants.TOPLINK_SERVICEID_PREFIX)
.append(ipAddress).append(":").append(port);
return sb.toString();
}
public Member getLocalMember() {
Service service = namedCache.getCacheService();
Cluster cluster = service.getCluster();
Member member = cluster.getLocalMember();
return member;
}
}
5) CoherenceDiscoveryManager繼承DiscoveryManager，主要重寫了startDiscovery和stopDiscovery方法
1package com.oocl.isdc.sha.frm.tts.remotecommand;
2
3import oracle.toplink.exceptions.ValidationException;
4import oracle.toplink.remotecommand.DiscoveryManager;
5import oracle.toplink.remotecommand.RemoteCommandManager;
6
7public class CoherenceDiscoveryManager extends DiscoveryManager {
8
9 public CoherenceDiscoveryManager(RemoteCommandManager rcm) {
10 super(rcm);
11 }
12
13 public RemoteCommandManager getRemoteCommandManager() {
14 return this.rcm;
15 }
16
17 public boolean isDiscoveryStopped() {
18 throw ValidationException.operationNotSupported("isDiscoveryStopped");
19 }
20
21 public void startDiscovery() {
22 ((CoherenceTransportManager)rcm.getTransportManager()).createLocalConnection();
23 }
24
25 /** *//**
26 * We must implement this method, and we keep it blank
27 */
28 public void stopDiscovery() {
29
30 }
31
32 public void setAnnouncementDelay(int millisecondsToDelay) {
33 throw ValidationException.operationNotSupported("setAnnouncementDelay");
34 }
35
36 public int getAnnouncementDelay() {
37 throw ValidationException.operationNotSupported("getAnnouncementDelay");
38 }
39
40 public String getMulticastGroupAddress() {
41 throw ValidationException.operationNotSupported("getMulticastGroupAddress");
42 }
43
44 public void setMulticastGroupAddress(String address) {
45 throw ValidationException.operationNotSupported("setMulticastGroupAddress");
46 }
47
48
49 public void setMulticastPort(int port) {
50 throw ValidationException.operationNotSupported("setMulticastPort");
51 }
52
53 public int getMulticastPort() {
54 throw ValidationException.operationNotSupported("getMulticastPort");
55 }
56}
57
3 命令行下mvn clean install部署Demo，部署的oc4j instance信息需要在ear下的pom.xml中給出。
<properties>
<home.j2ee>D:/oc4j_extended_101310/j2ee/home</home.j2ee>
<oc4j.host>localhost</oc4j.host>
<rmi.port>23791</rmi.port>
<deploy.username>oc4jadmin</deploy.username>
<deploy.password>welcome</deploy.password>
</properties>
因為我們需要檢測cache同步還需要部署到另外一個oc4j instance上，需要修改build.properties為：
coherence.member.name = tts-server1
coherence.local.address = 146.222.51.20
coherence.local.port = 8088
ear下的pom.xml為：
<properties>
<home.j2ee>C:/oc4j_extended_101310/j2ee/home</home.j2ee>
<oc4j.host>localhost</oc4j.host>
<rmi.port>23792</rmi.port>
<deploy.username>oc4jadmin</deploy.username>
<deploy.password>welcome</deploy.password>
</properties>
4 通過瀏覽器訪問demo，并檢測cache實現，我們在一個oc4j instance上修改employee的數據在另外一個oc4j的instance中就會立即呈現這個改變。并可以看到merge employee的相關toplink的log。如果沒有cache同步（demo的cache和cache 同步策略都基本采用了toplink的默認配置），因為employee上有樂觀鎖，當在另外一個oc4j instance或者說服務器上修改employee的數據，就會出現樂觀鎖異常，會rollback此次修改，如果有cache同步就會拿到最新的數據而不需要直接訪問db，最新的數據和db一致，大大減少了樂觀鎖出現的頻率。當然為了輔助更好的使用cache同步我們還需要定義cache invalidation機制。當然還有很多其他的cache策略避免出現臟數據。
Toplink Log：
[TopLink 非常詳細]: 2008.05.23 07:50:16.355--ServerSession(7674162)--Thread(Thread[DistributedCache:EventDispatcher,5,Cluster])--Received remote command oracle.toplink.remotecommand.MergeChangeSetCommand from Service[OOCL Toplink Coherence, 26464827, tcmp://146.222.51.20:8089]
[TopLink 非常詳細]: 2008.05.23 07:50:16.355--ServerSession(7674162)--Thread(Thread[DistributedCache:EventDispatcher,5,Cluster])--Executing command oracle.toplink.remotecommand.MergeChangeSetCommand from Service[OOCL Toplink Coherence, 26464827, tcmp://146.222.51.20:8089]
[TopLink 較詳細]: 2008.05.23 07:50:16.355--ServerSession(7674162)--Thread(Thread[DistributedCache:EventDispatcher,5,Cluster])--Received updates from Remote Server
[TopLink 非常詳細]: 2008.05.23 07:50:16.355--ServerSession(7674162)--Thread(Thread[DistributedCache:EventDispatcher,5,Cluster])--Merging com.oocl.isdc.sha.frm.tts.model.Employee: [558] from remote server
頁面演示：
上面兩個網頁截圖上的url表明是兩個不同的oc4j server。并且他們用了同一個db也是同一個application。在在server1上修改數據：
