JGroups简介和例子
柳宪
JGroups是一个组播通信工具，它可以：
[list]
[*]创建和删除一个组
[*]加入和离开某个组
[*]管理组成员关系，当有新的成员进入或存在的成员离开的时候会通知组内其它成员
[*]侦测和移除出现故障的组成员
[*]发送单播消息（unicast，point-to-point）
[*]发送广播消息（multicast，point-to-multipoint）
[/list]
JGroups的强大之处在于它有一个很灵活的协议栈，可以根据你的需要，随意的添加或删除某些功能。比如说，刚开始你使用IP广播发送你的消息，过了一会程序开始要求无损的消息传输，你可以添加NAKACK协议，它能确保接收方一定能收到你发送的消息。但是此时接收方收到的消息的顺序是不固定的，为了让接收顺序和发送顺序保持一致，你可以选择添加FIFO协议来确保一对收发者之间发送和接收的顺序。如果要确保组里所有成员的收发顺序，你可以添加TOTAL协议。再接下来，你可以添加GMS和FLUSH协议来维护组成员间的关系；FD协议可以进行故障检测；STATE_TRANSFER协议可以让新加入的组成员从已存在的成员中获取一致的状态；最后你还可以使用CRYPT协议来加密你发送的消息。
下面开始演示一个聊天组的程序。我们建立一个聊天组，分别发送单播消息和广播消息，当组成员发生变化的时候，所有组成员自动获得新的组成员视图，每当聊天组中加入一个新的成员的时候，新成员先和已存在的组成员进行状态同步（获取聊天记录）。
可以看到JGroups提供的API屏蔽了底层的通信机制，对于开发人员来说是完全透明的，要关注的只是消息的接受处理就可以了。
对应的协议栈配置：
[list]
[*]创建和删除一个组
[*]加入和离开某个组
[*]管理组成员关系，当有新的成员进入或存在的成员离开的时候会通知组内其它成员
[*]侦测和移除出现故障的组成员
[*]发送单播消息（unicast，point-to-point）
[*]发送广播消息（multicast，point-to-multipoint）
[/list]
JGroups的强大之处在于它有一个很灵活的协议栈，可以根据你的需要，随意的添加或删除某些功能。比如说，刚开始你使用IP广播发送你的消息，过了一会程序开始要求无损的消息传输，你可以添加NAKACK协议，它能确保接收方一定能收到你发送的消息。但是此时接收方收到的消息的顺序是不固定的，为了让接收顺序和发送顺序保持一致，你可以选择添加FIFO协议来确保一对收发者之间发送和接收的顺序。如果要确保组里所有成员的收发顺序，你可以添加TOTAL协议。再接下来，你可以添加GMS和FLUSH协议来维护组成员间的关系；FD协议可以进行故障检测；STATE_TRANSFER协议可以让新加入的组成员从已存在的成员中获取一致的状态；最后你还可以使用CRYPT协议来加密你发送的消息。
下面开始演示一个聊天组的程序。我们建立一个聊天组，分别发送单播消息和广播消息，当组成员发生变化的时候，所有组成员自动获得新的组成员视图，每当聊天组中加入一个新的成员的时候，新成员先和已存在的组成员进行状态同步（获取聊天记录）。
public class GroupChat extends ExtendedReceiverAdapter{
private JChannel channel;
private List<String> msgList = new ArrayList<String>(); //模拟状态对象，保存的是本节点的收到的消息
/**
* 在有节点向本节点请求状态的时候被调用
*/
@Override
public byte[] getState() {
byte[] state = null;
synchronized(msgList){
try {
state = Util.objectToByteBuffer(msgList);
} catch (Exception e) {
e.printStackTrace();
}
return state;
}
}
/**
* 在其他节点返回状态给本节点的时候被调用
*/
@Override
public void setState(byte[] state) {
synchronized(msgList){
try {
List<String> tmpList = (List<String>)Util.objectFromByteBuffer(state);
msgList.clear();
msgList.addAll(tmpList);
System.err.println("===receive state:[");
for(String msg : msgList){
System.out.println(msg);
}
System.out.println("]");
} catch (Exception e) {
e.printStackTrace();
}
}
}
/**
* 当有消息进来的时候被调用
*/
@Override
public void receive(Message msg) {
System.out.println(">>>new message receive from "+msg.getSrc()+":"+msg.getObject());
msgList.add(msg.getSrc()+" send : "+ msg.getObject());
}
/**
* 当组成员发生变化的时候被调用
*/
@Override
public void viewAccepted(View new_view) {
System.out.println("***new view receiver:"+new_view);
}
public void start() throws ChannelException{
//打开channel并指定配置文件
channel = new JChannel("udp.xml");
//指定组名，就可以创建或连接到广播组
channel.connect("ChatGroup");
//注册回调接口，使用"推"模式来接受广播信息
channel.setReceiver(this);
//查看当前组成员
System.out.println("---current view:"+channel.getView());
//状态同步，第一个参数为null则向协调者节点获取信息
channel.getState(null, 1000);
}
public void close(){
channel.close();
}
public void loopSendMessage(){
BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
try {
while(true){
System.out.println("please input a string");
String line = br.readLine();
System.out.println("you put:"+line);
if(line.equals("exit")){
break;
}else{
String[] array = line.split(",");
Address des = null; //接收方地址，为null代表发送广播消息
Address src = null; //发送方地址，为null代表自己的地址
String msg = line; //发送内容
if(array.length == 3){
des = new IpAddress(Integer.parseInt(array[0]));
src = new IpAddress(Integer.parseInt(array[1]));
msg = array[2];
}else if(array.length == 2){
des = new IpAddress(Integer.parseInt(array[0]));
msg = array[1];
}
Message message = new Message(des, src, msg);
//发送消息
channel.send(message);
}
}
} catch (IOException e) {
e.printStackTrace();
} catch (ChannelNotConnectedException e) {
e.printStackTrace();
} catch (ChannelClosedException e) {
e.printStackTrace();
}
}
public static void main(String[] args) {
GroupChat chat = new GroupChat();
try {
chat.start();
chat.loopSendMessage();
chat.close();
} catch (ChannelException e) {
e.printStackTrace();
}
}
}
可以看到JGroups提供的API屏蔽了底层的通信机制，对于开发人员来说是完全透明的，要关注的只是消息的接受处理就可以了。
对应的协议栈配置：
<config>
<UDP
mcast_group_addr="${jgroups.udp.mcast_addr:228.10.10.10}"
mcast_port="${jgroups.udp.mcast_port:45588}"
tos="8"
ucast_recv_buf_size="20000000"
ucast_send_buf_size="640000"
mcast_recv_buf_size="25000000"
mcast_send_buf_size="640000"
loopback="false"
discard_incompatible_packets="true"
max_bundle_size="64000"
max_bundle_timeout="30"
use_incoming_packet_handler="true"
ip_ttl="${jgroups.udp.ip_ttl:2}"
enable_bundling="true"
enable_diagnostics="true"
thread_naming_pattern="cl"
use_concurrent_stack="true"
thread_pool.enabled="true"
thread_pool.min_threads="1"
thread_pool.max_threads="25"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="false"
thread_pool.queue_max_size="100"
thread_pool.rejection_policy="Run"
oob_thread_pool.enabled="true"
oob_thread_pool.min_threads="1"
oob_thread_pool.max_threads="8"
oob_thread_pool.keep_alive_time="5000"
oob_thread_pool.queue_enabled="false"
oob_thread_pool.queue_max_size="100"
oob_thread_pool.rejection_policy="Run"/>
<PING timeout="2000"
num_initial_members="3"/>
<MERGE2 max_interval="30000"
min_interval="10000"/>
<FD_SOCK/>
<FD timeout="10000" max_tries="5" shun="true"/>
<VERIFY_SUSPECT timeout="1500" />
<BARRIER />
<pbcast.NAKACK max_xmit_size="60000"
use_mcast_xmit="false" gc_lag="0"
retransmit_timeout="300,600,1200,2400,4800"
discard_delivered_msgs="true"/>
<UNICAST timeout="300,600,1200,2400,3600"/>
<pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="400000"/>
<VIEW_SYNC avg_send_interval="60000" />
<pbcast.GMS print_local_addr="true" join_timeout="3000"
join_retry_timeout="2000" shun="false"
view_bundling="true"/>
<FC max_credits="20000000"
min_threshold="0.10"/>
<FRAG2 frag_size="60000" />
<!--pbcast.STREAMING_STATE_TRANSFER /-->
<pbcast.STATE_TRANSFER />
<!-- pbcast.FLUSH /-->
</config>
类似资料：