用Oracle Coherence实现Toplink Session Cache之间的数据同步(1)

ldd600

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>toplinkSyncache-name>
           scheme-name>DistributedCacheSchemescheme-name>
       cache-mapping>
    caching-scheme-mapping>

    caching-schemes>
       distributed-scheme>
           scheme-name>DistributedCacheSchemescheme-name>
           service-name>DistributedCacheservice-name>
           backing-map-scheme>
              local-scheme>
                  scheme-ref>DistributedMapscheme-ref>
              local-scheme>
           backing-map-scheme>
           backup-count>0backup-count>
           autostart>trueautostart>
       distributed-scheme>
       local-scheme>
           scheme-name>DistributedMapscheme-name>
           eviction-policy>LRUeviction-policy>
           high-units>10000high-units>
           expiry-delay>1Dexpiry-delay>
           flush-delay>1Dflush-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 serversrole-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.0from-address>
              to-address>146.222.51.255to-address>
           host-range>
       authorized-hosts>
       packet-publisher>
           packet-delivery>
              timeout-milliseconds>30000timeout-milliseconds>
           packet-delivery>
       packet-publisher>
    cluster-config>

    logging-config>
       destination>stderrdestination>
       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 = OOCL Toplink Coherence

第一个参数用来根据不同的情况使用不同toplink session的配置文件。第二个参数是toplink Command Channel的名字，唯一标识一个toplink cluster。

5          主要代码
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相关的连接。
package com.oocl.isdc.sha.frm.tts.remotecommand;

import oracle.toplink.exceptions.CommunicationException;
import oracle.toplink.internal.remotecommand.RemoteConnection;
import oracle.toplink.remotecommand.Command;
import oracle.toplink.remotecommand.RemoteCommandManager;

import com.oocl.isdc.sha.frm.tts.cohererence.cache.CoherenceCache;
import com.tangosol.util.MapEvent;
import com.tangosol.util.MapListener;

public class CoherenceRemoteConnection extends RemoteConnection implements MapListener {

    /** Comment for serialVersionUID */
    private static final long serialVersionUID = 8527315103990557963L;

    private CoherenceCache cache;

    private RemoteCommandManager rcm;

    public CoherenceRemoteConnection(RemoteCommandManager rcm, CoherenceCache cache) {
        this.serviceId = rcm.getServiceId();
        this.rcm = rcm;
        this.cache = cache;
    }

    /**
     * When some object in toplink session cache is chaged, it will callback this
     * method to put the changed object infomation to coherence cache
     */
    public Object executeCommand(Command command) throws CommunicationException {
        cache.putCache(command.getServiceId(), command);
        return null;
    }

    @SuppressWarnings("unchecked")
    protected void processObject(Object object) {
        Command command = null;
        if (object instanceof Command) {
            command = (Command) object;
            if (command.getServiceId().getChannel().equals(serviceId.getChannel())) {
                rcm.processCommandFromRemoteConnection(command);
            }
        } else if (null == object) {

        } else {

        }

    }

    public void close() {
        this.cache.removeMapListener(this);
    }
   
    public void becomeMapListener() {
        this.cache.addMapListener(this);
    }

    public void entryDeleted(MapEvent arg0) {
    }

    /**
     * When an object is inserted into coherence, this method of
     * listener will be called
     */
    public void entryInserted(MapEvent event) {
        processObject(event.getNewValue());
    }

    /**
     * When an object in coherence cache is updated, this method
     * of listener will be called
     */
    public void entryUpdated(MapEvent event) {
        processObject(event.getNewValue());
    }

}



本文来自ChinaUnix博客，如果查看原文请点：http://blog.chinaunix.net/u/19897/showart_702905.html