EhCache RMI 分布式緩存/緩存集群

EhCache 系統簡介

EhCache 是一個純 Java 的進程內緩存框架，具有快速、精干等特點。 
EhCache 的主要特性有：

1. 快速、精干
2. 簡單；
3. 多種緩存策略；
4. 緩存數據有兩級：內存和磁盤，因此無需擔心容量問題；
5. 緩存數據會在虛擬機重啟的過程中寫入磁盤；
6. 可以通過 RMI、可插入 API 等方式進行分布式緩存；
7. 具有緩存和緩存管理器的偵聽接口；
8. 支持多緩存管理器實例，以及一個實例的多個緩存區域；
9. 提供 Hibernate 的緩存實現；

EhCache集群解決的問題： 
　　由 於 EhCache 是進程中的緩存系統，一旦將應用部署在集群環境中，每一個節點維護各自的緩存數據，當某個節點對緩存數據進行更新，這些更新的數據無法在其它節點中共享， 這不僅會降低節點運行的效率，而且會導致數據不同步的情況發生。例如某個網站采用 A、B 兩個節點作為集群部署，當 A 節點的緩存更新后，而 B 節點緩存尚未更新就可能出現用戶在瀏覽頁面的時候，一會是更新后的數據，一會是尚未更新的數據。 
　　所以就需要用到 EhCache 的集群解決方案。 
　　 
EhCache集群方案：

• Terracotta 
• RMI 
• JMS : 依賴 ehcache-jmsreplication.jar 
• JGroups : 依賴ehcache-jgroupsreplication.jar 
• EhCache Server

　　其中的三種最為常用集群方式，分別是 RMI、JGroups 以及 EhCache Server 。

EhCache集群疑問

• 你如何知道集群環境中的其他緩存？ 
• 分布式傳送的消息是什么形式？ 
• 什么情況需要進行復制？增加（Puts），更新（Updates）或是失效（Expiries）？ 
• 采用什么方式進行復制？同步還是異步方式？

EhCache集群基本概念

1、正確的元素類型：只有可序列化的元素可以進行復制。一些操作，比如移除，只需要元素的鍵值而不用整個元素；在這樣的操作中即使元素不是可序列化的但鍵值是可序列化的也可以被復制。 
2、成員發現（Peer Discovery）：Ehcache進行集群的時候有一個cache組的概念。每個cache都是其他cache的一個peer，沒有主cache的存在。成員發現（Peer Discovery）正是用來解決 “你如何知道集群環境中的其他緩存?” 這個問題的。Ehcache提供了兩種機制用來進行成員發現，即：自動成員發現和手動成員發現。要使用一個內置的成員發現機制要在ehcache的配置文件中指定cacheManagerPeerProviderFactory元素的class屬性為 
net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory。

　　自動的發現方式用TCP廣播機制來確定和維持一個廣播組。它只需要一個簡單的配置可以自動的在組中添加和移除成員。在集群中也不需要什么優化服務器的知識，這是默認推薦的。ehcache.xml配置示例代碼如下：

<cacheManagerPeerProviderFactory
class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
properties="peerDiscovery=automatic, multicastGroupAddress=230.0.0.1,
multicastGroupPort=4446, timeToLive=32"/>
<!-- timeToLive
    0是限制在同一個服務器
    1是限制在同一個子網
    32是限制在同一個網站
    64是限制在同一個region
    128是限制在同一個大洲
    255是不限制-->

　手動的發現方式需要知道每個監聽器的IP地址和端口。集群成員（也就是服務器）不能在運行時動態地添加和移除。ehcache.xml配置示例代碼如下：

<!-- server1 -->
<cacheManagerPeerProviderFactory
class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
properties="peerDiscovery=manual,
rmiUrls=//server2:40001/sampleCache11|//server2:40001/sampleCache12"/>

<!-- server2 -->
<cacheManagerPeerProviderFactory
class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
properties="peerDiscovery=manual,
rmiUrls=//server1:40001/sampleCache11|//server1:40001/sampleCache12"/>

 

CacheManagerPeerListener 
　　每個CacheManagerPeerListener監聽從成員們發向當前CacheManager的消息。配置 CacheManagerPeerListener需要指定一個CacheManagerPeerListenerFactory，它以插件的機制實現， 用來創建CacheManagerPeerListener。 
　　Ehcache有一個內置的基於RMI的分布系統。它的監聽器是RMICacheManagerPeerListener，這個監聽器可以用RMICacheManagerPeerListenerFactory來配置。 
　　示例代碼：

<cacheManagerPeerListenerFactory
class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"
properties="hostName=localhost, port=40001,
socketTimeoutMillis=2000"/>

　　屬性說明：

hostname (可選) – 運行監聽器的服務器名稱。標明了做為集群群組的成員的地址，同時也是你想要控制的從集群中接收消息的接口。 
　　 在CacheManager初始化的時候會檢查hostname是否可用。 
　　 如果hostName不可用，CacheManager將拒絕啟動並拋出一個連接被拒絕的異常。 
　　如果指定，hostname將用InetAddress.getLocalHost().getHostAddress()來得到。 
port – 監聽器監聽的端口。 
socketTimeoutMillis (可選) – Socket超時的時間。默認是2000ms。當你socket同步緩存請求地址比較遠，不是本地局域網。你可能需要把這個時間配置大些，不然很可能延時導致同步緩存失敗。

CacheReplicators 
　　每個要進行同步的cache都需要設置一個用來向CacheManager的成員復制消息的緩存事件監聽器。這個工作要通過為每個cache的配置增加一個cacheEventListenerFactory元素來完成。 
　　代碼:

<cache name="sampleCache2" maxElementsInMemory="10" eternal="false" timeToIdleSeconds="100" timeToLiveSeconds="100" overflowToDisk="false">
<cacheEventListenerFactory class="net.sf.ehcache.distribution.RMICacheReplicatorFactory"
properties="replicateAsynchronously=true,replicatePuts=true, replicateUpdates=true, replicateUpdatesViaCopy=false, replicateRemovals=true "/>
</cache>

　　cacheEventListenerFactory 支持以下屬性

replicatePuts=true | false – 當一個新元素增加到緩存中的時候是否要復制到其他的peers. 默認是true。 
replicateUpdates=true | false – 當一個已經在緩存中存在的元素被覆蓋時是否要進行復制。默認是true。 
replicateRemovals= true | false – 當元素移除的時候是否進行復制。默認是true。 
replicateAsynchronously=true | false – 復制方式是異步的（指定為true時）還是同步的（指定為false時）。默認是true。 
replicatePutsViaCopy=true | false – 當一個新增元素被拷貝到其他的cache中時是否進行復制指定為true時為復制，默認是true。

Cache屬性說明：

<cache 
        name="userCache" 
        maxElementsInMemory="10000" 
        eternal="true" 
        overflowToDisk="true" 
        timeToIdleSeconds="0" 
        timeToLiveSeconds="0"
        diskPersistent="false" 
        diskExpiryThreadIntervalSeconds="120">
        <cacheEventListenerFactory 
         　　class="net.sf.ehcache.distribution.RMICacheReplicatorFactory" 
     　　 　　properties="replicateAsynchronously=true,replicatePuts=true,replicateUpdates=true,replicateUpdatesViaCopy=false,replicateRemovals=true">  
        </cacheEventListenerFactory>
        <bootstrapCacheLoaderFactory
     　　 　　class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory"
             　　properties="bootstrapAsynchronously=true"> 
        </bootstrapCacheLoaderFactory>
    </cache>

 

必須屬性： 
　　name:緩存名稱。 
　　maxElementsInMemory：緩存最大個數。 
　　eternal:對象是否永久有效，一但設置了，timeout將不起作用。 
　　overflowToDisk：當內存中對象數量達 
　　maxElementsInMemory時，Ehcache將會對象寫到磁盤中。 
　　 diskSpoolBufferSizeMB：這個參數設置DiskStore（磁盤緩存）的緩存區大小。默認是30MB。每個Cache都應該有自己的一個緩沖區。 
　　maxElementsOnDisk：硬盤最大緩存個數。 
可選的屬性： 
　　timeToIdleSeconds：設置對象在失效前的允許閑置時間（單位：秒）。僅當eternal=false對象不是永久有效時使用，可選屬性，默認值是0，也就是可閑置時間無窮大。 
　　timeToLiveSeconds：設置對象在失效前允許存活時間（單位：秒）。最大時間介於創建時間和失效時間之間。僅當eternal=false對象不是永久有效時使用，默認是0.，也就是對象存活時間無窮大。 
　　diskPersistent：是否disk store在虛擬機啟動時持久化. The default value is false. 
　　memoryStoreEvictionPolicy：當達到maxElementsInMemory限制時，Ehcache將會根據指定的策略去清理內存。默認策略是LRU（最近最少使用）。你可以設置為FIFO（先進先出）或是LFU（較少使用）。 
　　diskExpiryThreadIntervalSeconds：磁盤失效線程運行時間間隔，默認是120秒。 
　　clearOnFlush：內存數量最大時是否清除。 
緩存子元素： 
　　cacheEventListenerFactory：注冊相應的的緩存監聽類，用於處理緩存事件，如put,remove,update,和expire 
　　bootstrapCacheLoaderFactory:指定相應的BootstrapCacheLoader，用於在初始化緩存，以及自動設置。

EhCache RMI 手動方式配置緩存

Tomcat1: 127.0.0.1:8080, Cache Server1: 127.0.0.1:40001 
Tomcat2: 127.0.0.1:8088, Cache Server2: 127.0.0.1:40002

用到的架包： 
ehcache-core-2.6.8.jar 
log4j-1.2.17.jar 
servlet-api.jar (示例中使用了@WebServlet，請確保servler-api的架包版本大於3.0) 
slf4j-api-1.6.1.jar 
slf4j-log4j12-1.6.1.jar

創建Web工程TestEhcache， 工程目錄如下： 

CacheManagerFactory.java

private CacheManager manager;
    private static CacheManagerFactory factory = new CacheManagerFactory();
    private final static String EHCACHEFILE = "/ehcache.xml";

    private CacheManagerFactory() {
    }

    public static CacheManagerFactory getInstance() {
        return factory;
    }

    public CacheManager getCacheManager() {
        if (manager == null) {
            InputStream is = this.getClass().getResourceAsStream(EHCACHEFILE);
            manager = CacheManager.create(is);
        }
        return manager;
    }

    public Cache getCache(String cache) {
        return getCacheManager().getCache(cache);
    }

    public void setCache(Cache cache) {
        getCacheManager().addCache(cache);
    }

    public void setCache(String cache) {
        getCacheManager().addCache(cache);
    }

    public Element getElement(String cacheName, String key) {
        if (getCache(cacheName) == null)
            setCache(cacheName);
        return getCache(cacheName).get(key);
    }

    public void setElement(String cache, Element element) {
        if (getCache(cache) == null)
            setCache(cache);
        getCache(cache).put(element);
    }

    public Boolean continaElementKey(String cacheName, String key) {
        if (getCache(cacheName) == null)
            setCache(cacheName);
        return getCache(cacheName).isKeyInCache(key);
    }

 

TesAction.java

@WebServlet("/test")
public class TesAction extends HttpServlet {

    private static final long serialVersionUID = 1L;

    CacheManagerFactory cmf = CacheManagerFactory.getInstance();

    protected void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
        doPost(request,response);
    }

    protected void doPost(HttpServletRequest request, HttpServletResponse response) throws IOException {
        String res = "";
        String key = request.getParameter("key");

        Element element = cmf.getElement("userCache", "map");
        if(element == null){
            Map<String, String> map = new HashMap<String, String>();
            map.put(key, key);
            cmf.setElement("userCache", new Element("map", map));
        }else{
            Map<String, String> map = (Map<String, String>) element.getValue();
            res = map.get(key);
            if(res == null){
                map.put(key, key);
                // 多次測試發現，存在同名Element是，重復put的是無法復制的，因此當遇到兩個節點同步不上的時候，先remove后put。 
                cmf.getCache("userCache").remove("map");
                cmf.setElement("userCache", new Element("map", map));
                res = "0;null";
            }
        }

        response.setContentType("text/html;charset=UTF-8");
        response.setCharacterEncoding("UTF-8");
        PrintWriter out = response.getWriter();
        out.write(res);
        out.close();
    }

    @Override
    public void init() throws ServletException {
        super.init();
    }

}

 

ehcache.xml:

<?xml version="1.0" encoding="UTF-8"?>
<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://ehcache.sf.net/ehcache.xsd">

    <diskStore path="java.io.tmpdir" />

    <cacheManagerPeerProviderFactory 
        class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
        properties="peerDiscovery=manual,rmiUrls=//127.0.0.1:40002/userCache|//127.0.0.1:40002/resourceCache">
    </cacheManagerPeerProviderFactory>

    <cacheManagerPeerListenerFactory 
        class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"
        properties="hostName=127.0.0.1, port=40001, socketTimeoutMillis=2000">
    </cacheManagerPeerListenerFactory>

    <defaultCache maxElementsInMemory="10000" eternal="false" timeToIdleSeconds="30" timeToLiveSeconds="30" overflowToDisk="false"/>

    <cache 
        name="userCache" 
        maxElementsInMemory="10000" 
        eternal="true" 
        overflowToDisk="true" 
        timeToIdleSeconds="0" 
        timeToLiveSeconds="0"
        diskPersistent="false" 
        diskExpiryThreadIntervalSeconds="120">
        <cacheEventListenerFactory 
            class="net.sf.ehcache.distribution.RMICacheReplicatorFactory" 
            properties="replicateAsynchronously=true,replicatePuts=true,replicateUpdates=true,replicateUpdatesViaCopy=false,replicateRemovals=true">    
        </cacheEventListenerFactory>
        <bootstrapCacheLoaderFactory
            class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory"
            properties="bootstrapAsynchronously=true"> 
        </bootstrapCacheLoaderFactory>
    </cache>

    <cache 
        name="resourceCache" 
        maxElementsInMemory="10000" 
        eternal="true" 
        overflowToDisk="true" 
        timeToIdleSeconds="0" 
        timeToLiveSeconds="0"
        diskPersistent="false" 
        diskExpiryThreadIntervalSeconds="120">
        <cacheEventListenerFactory 
            class="net.sf.ehcache.distribution.RMICacheReplicatorFactory" 
            properties="replicateAsynchronously=true,replicatePuts=true,replicateUpdates=true,replicateUpdatesViaCopy=false,replicateRemovals=true">    
        </cacheEventListenerFactory>
        <bootstrapCacheLoaderFactory
            class="net.sf.ehcache.distribution.RMIBootstrapCacheLoaderFactory"
            properties="bootstrapAsynchronously=true"> 
        </bootstrapCacheLoaderFactory>
    </cache>

</ehcache>

 

復制TestEhcache到 TestEhcache1, 修改TestEhcache1下的ehcache.xml ，只需要修改cacheManagerPeerProviderFactory和cacheManagerPeerListenerFactory修改為如下代碼，其他不變

<cacheManagerPeerProviderFactory 
        class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
        properties="peerDiscovery=manual,rmiUrls=//127.0.0.1:40001/userCache|//127.0.0.1:40001/resourceCache">
    </cacheManagerPeerProviderFactory>

    <cacheManagerPeerListenerFactory 
        class="net.sf.ehcache.distribution.RMICacheManagerPeerListenerFactory"
        properties="hostName=127.0.0.1, port=40002, socketTimeoutMillis=2000">
    </cacheManagerPeerListenerFactory>

為了區別，將TestEhcache1的TesAction.doPost()方法修改為如下代碼，只get的緩存不put緩存，用於觀察TestEhcache1是否能同步TestEhcache的數據：

String res = "0;null";
        Element element = cmf.getElement("userCache", "map");
        if(element != null){
            Map<String, String> map = (Map<String, String>) element.getValue();
            res = map.get(request.getParameter("key"));
            if(res == null){
                res = "0;null";
            }
        }
        response.setContentType("text/html;charset=UTF-8");
        response.setCharacterEncoding("UTF-8");
        PrintWriter out = response.getWriter();
        out.write(res);
        out.close();

將TestEhcache部署到tomcat1, TestEhcache1部署到Tomcat2。依次執行如下代碼： 
localhost:8080/TestEhcache/test?key=125 
localhost:8080/TestEhcache/test?key=125 
localhost:8088/TestEhcache1/test?key=125 
如果輸出內容分別如下，說明集群ok, 兩節點數據同步沒問題， 否者請仔細檢查配置文件和TestAction代碼： 
0;null 
125 
125

EhCache RMI 自動方式配置緩存 
將ehcache.xml的cacheManagerPeerProviderFactory代碼改為：

<cacheManagerPeerProviderFactory
 class="net.sf.ehcache.distribution.RMICacheManagerPeerProviderFactory"
    properties="peerDiscovery=automatic, multicastGroupAddress=192.168.0.1,
    multicastGroupPort=40004, timeToLive=32"
/>

EhCache Jgroups 方式配置緩存

在TestEhcache和TestEhcache1工程中添加ehcache-jgroupsreplication-1.7.jar和jgroups-3.6.9.Final.jar。 使用該方式比RMI方式配置簡單。

TestEhcache 的 ehcache.xml

<?xml version="1.0" encoding="UTF-8"?>
<ehcache xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:noNamespaceSchemaLocation="http://ehcache.sf.net/ehcache.xsd">

    <diskStore path="java.io.tmpdir" />

    <cacheManagerPeerProviderFactory 
        class="net.sf.ehcache.distribution.jgroups.JGroupsCacheManagerPeerProviderFactory"
        properties="connect=TCP(bind_port=4001):
            TCPPING(initial_hosts=192.168.8.150[4001],192.168.8.150[4002];port_range=10;timeout=3000;num_initial_members=3):
            MERGE2(min_interval=3000;max_interval=5000):  
            FD_ALL(interval=5000;timeout=20000):  
            FD(timeout=5000;max_tries=48;):  
            VERIFY_SUSPECT(timeout=1500):  
            pbcast.NAKACK(retransmit_timeout=100,200,300,600,1200,2400,4800;discard_delivered_msgs=true):  
            pbcast.STABLE(stability_delay=1000;desired_avg_gossip=20000;max_bytes=0):  
            pbcast.GMS(print_local_addr=true;join_timeout=5000)"
        propertySeparator="::">
    </cacheManagerPeerProviderFactory>

    <defaultCache 
        maxElementsInMemory="10000" 
        eternal="true" 
        overflowToDisk="true" 
        timeToIdleSeconds="0" 
        timeToLiveSeconds="0"
        diskPersistent="false" 
        diskExpiryThreadIntervalSeconds="120">
        <cacheEventListenerFactory 
            class="net.sf.ehcache.distribution.jgroups.JGroupsCacheReplicatorFactory"
            properties="replicateAsynchronously=true,replicatePuts=true,replicateUpdates=true,replicateUpdatesViaCopy=false,replicateRemovals=true"/>
        <bootstrapCacheLoaderFactory
            class="net.sf.ehcache.distribution.jgroups.JGroupsBootstrapCacheLoaderFactory"
            properties="bootstrapAsynchronously=true"> 
        </bootstrapCacheLoaderFactory>
    </defaultCache>
</ehcache>

復制TestEhcache的ehcache.xml到TestEhcache1，並且修改下列代碼：

connect=TCP(bind_port=4002)

推薦文章：http://raychase.iteye.com/blog/1545906 
http://blog.csdn.net/tang06211015/article/details/52281551

http://www.ehcache.org/apidocs/2.10.3/index.html

 

來源：http://blog.csdn.net/xlxxcc/article/details/52350264