`

利用JGroups同步两台server之间的cache

Cachethread 
阅读更多

一、需求
前段时间做了一个项目，在后台有很多的数据都放入到了cache中了，而且还会对cache中的数据进行更新。如果只有一台server没有任何问题，但是如果考虑到集群负载平衡，连接多个server的时候，就有问题出现了，怎么样才能保证多个server之间cache的同步呢？请看下面的部署图。

二、引入JGroups
JGroups是一个可靠的组间通讯工具，进程可以加入一个通讯组，给组内所有的成员或单独的成员发送消息，同样，也可以从组中的成员处接收消息。
系统会记录组的每一个成员，在新成员加入或是现有的成员离开或是崩溃时，会通知组内的其他成员。

当我们更新一台server上的cache的时候，利用JGroups进行广播，其他的server接收到广播，根据接收到的信息来更新自己的cache，这样达到了
每个server的cache同步。

三、实现
1、定义一个接口BaseCache规定出对cache类操作的方法

 1public interface BaseCache {
 2    
 3    public void put(String key, Object ob);
 4    public Object get(String key);
 5    
 6    public void delete(String key);
 7    public void batchDelete(String[] list);
 8    public void batchDelete(List list);
 9    public void deleteAll();
10    
11}


2、定义一个同步器（CacheSynchronizer），这个类利用JGroups进行发送广播和接收广播

 1public class CacheSynchronizer {
 2    private static String protocolStackString=
 3        "UDP(mcast_addr=235.11.17.19;mcast_port=32767;ip_ttl=3;"+
 4        "mcast_send_buf_size=150000;mcast_recv_buf_size=80000):"+
 5        "PING(timeout=2000;num_initial_members=3):"+
 6        "MERGE2(min_interval=5000;max_interval=10000):"+
 7        "FD_SOCK:"+
 8        "VERIFY_SUSPECT(timeout=1500):"+
 9        "pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800):"+
10        "pbcast.STABLE(desired_avg_gossip=20000):"+
11        "UNICAST(timeout=2500,5000):"+
12        "FRAG:"+
13        "pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;shun=false;print_local_addr=false)";
14    private static String groupName="YHPTEST";
15    
16    private Channel jgroupsChannel=null;
17    
18    //inner class ，定义接收广播，已经对接收到的广播进行处理
19    private class ReceiveCallback extends ExtendedReceiverAdapter{
20        private BaseCache cache=null;
21        public void setCache(BaseCache baseCache){//设置cache类
22            cache=baseCache;
23        }
24        public void receive(Message msg) {
25            if(cache==null) return ;
26            String strMsg = (String) msg.getObject();
27            if(strMsg!=null&&(!"".equals(strMsg))){
28                cache.put(strMsg, strMsg);    //根据接收到的广播，同步cache
29            }
30        }
31    }
32    
33    private ReceiveCallback recvCallback = null;
34    
35    public CacheSynchronizer(BaseCache cache) throws Exception{
36        jgroupsChannel = new JChannel(protocolStackString);
37        recvCallback = new ReceiveCallback();
38        recvCallback.setCache(cache);
39        jgroupsChannel.setReceiver(recvCallback);
40        jgroupsChannel.connect(groupName);
41    }
42    
43    /** *//**
44     * 发送广播信息，我们可以自定义广播的格式。
45     * 这里简单起见，仅仅发送一个字符串
46     * @param sendMsg
47     * @throws Exception
48     */
49    public void sendCacheFlushMessage(String sendMsg)throws Exception {
50        jgroupsChannel.send(null, null, sendMsg); //发送广播
51        
52    }
53}


3、定义cache类，调用同步器同步cache

 1public class TestDataCache implements BaseCache {
 2    private Map dataCache=null;//保持cache数据
 3    private CacheSynchronizer cacheSyncer = null; //同步器
 4    
 5    //inner class for thread safe.
 6    private static final class TestDataCacheHold{
 7        private static TestDataCache  theSingleton=new TestDataCache();        
 8        public static TestDataCache getSingleton(){
 9            return theSingleton;
10        }
11        private TestDataCacheHold(){}
12    }
13    
14    //Prevents to inherit
15    private TestDataCache(){
16        dataCache=new HashMap();
17        createSynchronizer();
18    }
19    
20    public static TestDataCache getInstance(){
21        return TestDataCacheHold.getSingleton();
22    }
23    
24    public CacheSynchronizer getSynchronizer(){
25        return cacheSyncer;
26    }
27    
28    public int getCacheLength(){
29        return dataCache.size();
30    }
31    
32    public void createSynchronizer(){
33        try{
34            cacheSyncer=new CacheSynchronizer(this);
35        }catch(Exception e){
36            e.printStackTrace();
37        }
38    }
39    
40    public void batchDelete(String[] list) {
41        if(list!=null) return ;
42        synchronized (dataCache){
43            for(int i=0;i<list.length;i++){
44                if(list[i].length()>0){
45                    dataCache.remove(list[i]);
46                }
47            }
48        }
49
50    }
51http://www.blogjava.net/Images/OutliningIndicators/InBl
分享到：
| JGroups简介
评论
发表评论

您还没有登录,请您登录后再发表评论

相关推荐

    JavaEE源代码 jgroups-2.2.8

    JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 jgroups-2.2.8JavaEE源代码 ...

    Jgroups 教程

    通过这个通道，节点之间可以发送消息、同步状态等。 ##### 2.2 创建通道并加入群集 首先，需要创建一个JGroups通道，并指定配置文件。然后，调用`connect`方法加入到指定的群集中。 ##### 2.3 主事件循环与发送...

    Ehcache通过Jgroups做集群

    测试集群配置，可以启动多台虚拟机或实际的服务器，分别运行配置好的应用，然后观察Ehcache是否能够正确地在集群中同步数据。可以通过添加、更新或删除缓存项，然后检查其他节点是否也同步了这些变更。 在`...

    JGroups_集群.pdf

    JGroups已经被广泛应用于JBoss Application Server Clustering、OSCache Clustering、Jetty HTTP session replication、Tomcat HTTP session replication等领域。 JGroups的协议栈可以根据不同的需求进行定制，例如...

    jgroups-2.2.7.jar

    jgroups-2.2.7.jar jgroups-2.2.7.jar

    jgroups.part1

    jgroups.part1

    JGroups的Raft实现jgroups-raft.zip

    jgroups-raft 项目是 JGroups 框架对 Raft 的实现。Maven：&lt;groupId&gt;org.jgroups &lt;artifactId&gt;jgroups-raft &lt;version&gt;0.2&lt;/version&gt;Raft 是一个容易理解的共识算法。在容错和性能方面它相当于 Paxos（Google 的一致...

    jgroups源代码

    《深入解析JGroups开源框架：基于belaban-JGroups-19d7183源代码》 JGroups是一个用于构建高可用性集群的Java框架，它提供了可靠的消息传递、组成员管理和故障检测等功能，广泛应用于分布式系统中。本文将基于bela...

    jgroups-3.0.2

    JGroups是一个开源的纯java编写的可靠的群组通讯工具。其是一个可靠的组播通讯工具集（需要说明的是，这并不是说...JGroups 适合使用场合服务器集群cluster、多服务器通讯、服务器replication(复制)、分布式cache缓存等

    jgroups官方帮助文档html格式打包2.X版本

    此文档主要针对JGroups 2.X版本的官方帮助文档进行详细解读，旨在帮助开发者深入理解并有效地利用JGroups。 一、JGroups简介 JGroups的核心目标是确保在分布式环境中数据的一致性。它提供了一套完整的工具，用于...

    JGroups-jdk.zip_jgroups

    在本文中，我们将深入探讨JGroups的核心特性、工作原理以及如何利用它来创建可靠的分布式系统。 首先，JGroups的核心功能在于它的组通信能力。一个“组”是由多个成员组成的逻辑集合，每个成员都可以向组内的其他...

    jgroups

    ### 关于JGroups 2.5教程：安装与开发简易应用程序 #### 安装与配置JGroups **JGroups**是一款高性能、可扩展且高度可靠的群集通信库，旨在为分布式系统提供消息传递功能。本教程将深入探讨如何安装配置JGroups，...

    encache+jgroups集群缓存共享

    "encache+jgroups集群缓存共享"这个主题聚焦于如何利用EnCache和JGroups两个技术来实现高效的集群间缓存共享。 EnCache是一个高性能、分布式、内存中的键值存储系统，通常用于缓存应用程序的数据，以减少对数据库的...

    jgroups.part3

    jgroups.part3

    Jgroups中的UNICAST3协议中文翻译

    Jgroups 中的 UNICAST3 协议详解 Jgroups 是一种基于 IP 多播的可靠的组播中间件，UNICAST3 协议是 Jgroups 中的一种单播协议，旨在保持单播和 UNICAST2 的正面特征，而修正负面特征。 UNICAST3 协议的主要特点是...

    jgroups-3.2

    JGroups是一个开源的纯java编写的可靠的群组通讯工具。其是一个可靠的组播通讯工具集（需要说明的是，这并不是说必须要使用IP Multicast,JGroups也可以使用TCP来实现）。其工作模式基于IP多播，但可以在可靠性和群组...

    Ecahche+Jgroups

    Ehcache和JGroups是两个在分布式系统中广泛使用的开源技术。Ehcache是一个高性能、易用的Java本地缓存解决方案，而JGroups则是一个用于组通信的框架，专门处理集群中的节点间通信问题。当这两者结合时，可以构建出...

    Java多播通讯框架 JGroups

    Java多播通讯框架JGroups是Java开发者用于构建高可用、高性能和可伸缩的集群通信系统的重要工具。它提供了一套全面的协议栈，能够处理网络中的节点发现、消息传递、故障检测和恢复等问题，从而使得开发分布式应用变...

Magicbox
Global site tag (gtag.js) - Google Analytics