一、需求
前段时间做了一个项目，在后台有很多的数据都放入到了cache中了，而且还会对cache中的数据进行更新。如果只有一台server没有任何问题，但是如果考虑到集群负载平衡，连接多个server的时候，就有问题出现了，怎么样才能保证多个server之间cache的同步呢？请看下面的部署图。
二、引入JGroups
JGroups是一个可靠的组间通讯工具，进程可以加入一个通讯组，给组内所有的成员或单独的成员发送消息，同样，也可以从组中的成员处接收消息。
系统会记录组的每一个成员，在新成员加入或是现有的成员离开或是崩溃时，会通知组内的其他成员。
当我们更新一台server上的cache的时候，利用JGroups进行广播，其他的server接收到广播，根据接收到的信息来更新自己的cache，这样达到了
每个server的cache同步。
三、实现
1、定义一个接口BaseCache规定出对cache类操作的方法
public 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进行发送广播和接收广播
public 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); //根据接收到的广播，同步cache29
}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 sendMsg47
* @throws Exception48
*/49
public void sendCacheFlushMessage(String sendMsg)throws Exception {50
jgroupsChannel.send(null, null, sendMsg); //发送广播51
52
}53
}
3、定义cache类，调用同步器同步cache
public 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 inherit15
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
}51
52
public void batchDelete(List list) {53
synchronized (dataCache){54
Iterator itor=list.iterator();55
while(itor.hasNext()){56
String tmpKey=(String)itor.next();57
if(tmpKey.length()>0){58
dataCache.remove(tmpKey);59
}60
}61
}62
}63
64
public void delete(String key) {65
synchronized (dataCache) {66
dataCache.remove(key);67
}68
}69
70
public void deleteAll() {71
synchronized (dataCache){72
dataCache.clear();73
}74
75
}76
77
public Object get(String key) {78
Object theObj=null;79
synchronized (dataCache) {80
theObj =dataCache.get(key);81
}82
return theObj;83
}84
85
public void put(String key, Object obj) {86
Object theObj=null;87
synchronized (dataCache){88
theObj=dataCache.get(key);89
if(theObj==null){90
dataCache.put(key, obj);91
}else{92
theObj=obj;93
}94
}95
}96
97
}98
4、更新cache，测试是否同步
Scanner cin=new Scanner(System.in);2
String input=cin.next().trim();3
TestDataCache cache=TestDataCache.getInstance();4
while(!"q".equalsIgnoreCase(input)){ 5
if(!"".equals(input)){6
cache.put(input, input);7
cache.getSynchronizer().sendCacheFlushMessage(input);8
} 9
System.out.println(cache.getCacheLength());10
input=cin.next();11
}我打开两个Eclipse，run此程序，输入测试数据，控制台显示同步后cache的长度。
下面是相应的类图：
四、引申
在此实例中，我们为简单起见仅仅考虑了新增对cache的同步，如果是个真正的项目，这显然是不够的。这样我们就必须定义出消息的格式，例如操作的对象，操作的命令等等。根据消息的定义来执行同步数据的操作。
五、附录
Multicast是一种同时像多台机器发送数据的机制。
Multicast使用224.0.0.0 到 239.255.255.255 这段IP来传送数据，这段IP地址是保留的，发送到这上面的数据不会通过你的子网转发。
在RFC-1060中定义了一部分预留的组播地址，使用时应注意不要重复。
一些比较特别的组播地址：（更多内容请查看RFC-1060)
1) 224.0.0.0 这个是保留地址，不会被指定到任何的组播组
2) 224.0.0.1 这个地址在所有的主机上被指定为一个永久组播组，这个地址可以用来找到本地子网内所有的组播主机。
使用ping 224.0.0.1可以查看这些地址
在一个组播中的所有主机使用一个相同的组播地址，它们被称为一个组(Group),组中的成员是动态的，他们可以随时加入或者离开组。每台主机可以同时是多个组的成员，也可以不属于任何一个组。比较特别的是，并不是只有组中的成员才可以给组发送数据。
组分为两种，一种是永久性的，一种是动态的。对于永久性的组，他们拥有一个众所周知的管理IP地址，这个地址不是组中的成员，它是永久的。永久性的组可以拥有任何数量的成员，甚至没有成员。而动态组只有在拥有成员的时候才存在。JGroups使用的就是动态组来实现组播数据的。
六、参考
http://puras.javaeye.com/blog/81783
七、备注
在linux下，如果要调用JGroups，启动Tomcat时，必须修改catalina.sh，增加下面的参数：
JAVA_OPTS=" -Djava.net.preferIPv4Stack=true "
请参考下面的链接：
http://weblogs.java.net/blog/dcengija/archive/2006/04/jgroups_demos_o.html