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上修改数据：
