JGroups入门

  前些天看了些关于JGroups方面的资料，怕时间久了将之忘于脑后，觉得有必要记录一下，这样日后就不用再从头看啦。
    现在有很多项目都使用JGroups做底层的通讯，知道的开源项目有JBoss Cache和OSCache用它做为底层支持来实现集群的，一定还有其他的项目也用到了，只不过我不知道而已了。
    JGroups是一个可靠的组间通讯工具，进程可以加入一个通讯组，给组内所有的成员或单独的成员发送消息，同样，也可以从组中的成员处接收消息。系统会 记录组的每一个成员，在新成员加入或是现有的成员离开或是崩溃时，会通知组内的其他成员，这样我们就不必自己去管理这些事情了。
    要想加入一个组，并与组内其他的成员交互，必须建立一个Channel连接到组，同一个组内的所有成员使用相同的组名称。首先是创建一个Channel，可以直接实例化一个Channel的实现，这里用的是JChannel：

Java代码

1. JChannel channel =  new  JChannel(props);  

JChannel channel = new JChannel(props);

参数里指定Channel使用的协议栈，如果是空的，则使用默认的协议栈，位于JGroups包里的udp.xml。参数可以是一个以冒号分隔的字符串，或是一个XML文件，在XML文件里定义协议栈。
下面的是JGroups文档里给出的字符串的例子：

Java代码

1. String props= "UDP(mcast_addr=228.1.2.3;mcast_port=45566;ip_ttl=32):"  +  
2.         "PING(timeout=3000;num_initial_members=6):"  +  
3.         "FD(timeout=5000):"  +  
4.         "VERIFY_SUSPECT(timeout=1500):"  +  
5.         "pbcast.STABLE(desired_avg_gossip=10000):"  +  
6.         "pbcast.NAKACK(gc_lag=10;retransmit_timeout=3000):"  +  
7.         "UNICAST(timeout=5000;min_wait_time=2000):"  +  
8.         "FRAG:"  +  
9.         "pbcast.GMS(initial_mbrs_timeout=4000;join_timeout=5000;"  +  
10.         "join_retry_timeout=2000;shun=false;print_local_addr=false)" ;  
11. JChannel channel;  
12. try  {  
13.     channel=new  JChannel(props);  
14. }  
15. catch (Exception ex) {  
16. // channel creation failed   

String props="UDP(mcast_addr=228.1.2.3;mcast_port=45566;ip_ttl=32):" +
        "PING(timeout=3000;num_initial_members=6):" +
        "FD(timeout=5000):" +
        "VERIFY_SUSPECT(timeout=1500):" +
        "pbcast.STABLE(desired_avg_gossip=10000):" +
        "pbcast.NAKACK(gc_lag=10;retransmit_timeout=3000):" +
        "UNICAST(timeout=5000;min_wait_time=2000):" +
        "FRAG:" +
        "pbcast.GMS(initial_mbrs_timeout=4000;join_timeout=5000;" +
        "join_retry_timeout=2000;shun=false;print_local_addr=false)";
JChannel channel;
try {
	channel=new JChannel(props);
}
catch(Exception ex) {
// channel creation failed

    创建完之后，Channel现在处于未连接状态，需要通过connect方法将之连接到组，使其处于连接状态：

Java代码

1. public   void  connect(String groupname)  throws  ChannelClosed;  

public void connect(String groupname) throws ChannelClosed;

    它的参数就是要加入组的组名字，如果加入的组之前没有任何成员，则会自动创建一个组。
    此时，Channel已处于连接状态，可以发送/接收消息了，发送消息的方法为：

Java代码

1. public   void  send(Message msg)  throws  ChannelNotConnected, ChannelClosed;  
2. public   void  send(Address dst, Address src, Object obj)  throws  ChannelNotConnected, ChannelClosed;  

public void send(Message msg) throws ChannelNotConnected, ChannelClosed;
public void send(Address dst, Address src, Object obj) throws ChannelNotConnected, ChannelClosed;

    两个方法基本是一样的，只不过一个是直接提供一个消息，而另一个只是提供了消息的目的地，源，和消息内容，其实这个方法在内部也是通过第一个方法来实现 的，在其内部，将提供的这三个参数组合成一个消息，再调用第一个方法，具体使用哪个方法，则看个人喜好和实际情况了。
    消息的由消息的目的地，源，Flag，消息内容，Header组成。其中如果目的地为空，则认为是发给所有组成员的消息；源为空的话，在底层的协议将其放 到网络上时，会自动的将本Channel的地址填充进去。Address则是组成员的地址，用于唯一的标识一个组成员的接口，JGroups提供了几种默 认的实现。下面的例子是发送一条消息到组内所有的成员处：

Java代码

1. Hashtable data;  // any serializable data   
2. try  {  
3.     channel.send(null ,  null , data);  
4. }  
5. catch (Exception ex) {  
6. // handle errors   

Hashtable data; // any serializable data
try {
	channel.send(null, null, data);
}
catch(Exception ex) {
// handle errors

再来一个发送到单独的组成员的例子：

Java代码

1. Address receiver;  
2. Hashtable data;  
3. try  {  
4. receiver=channel.getView().getMembers().first();  
5. channel.send(receiver, null , data);  
6. }  
7. catch (Exception ex) {  
8. // handle errors   
9. }  

Address receiver;
Hashtable data;
try {
receiver=channel.getView().getMembers().first();
channel.send(receiver, null, data);
}
catch(Exception ex) {
// handle errors
}

    其中的channel.getView().getMembers().first()是指从Channel中取出当前的成员列表，再从中取出第一个成员。之后就可以将这个成员做为目的地来发送消息了。
    可以发送消息，同样也可以接收消息：

Java代码

1. public  Object receive( long  timeout)  throws  ChannelNotConnected, ChannelClosed, Timeout;  

public Object receive(long timeout) throws ChannelNotConnected, ChannelClosed, Timeout;

    利用此方法可以取回多种消息，如普通的消息，View消息，等等。它的timeout参数则是指定超时的时间，如果设置为0时，而此时又没有新消息可以接 收，此方法则会形成一个阻塞，在这一直等到有可用的消息为止；设置为大于0时，如果没有可用消息，超过此值后，会抛出一个Timeout异常。
    下面的列表则是可以接收的消息的详细清单：

Java代码

1. Message  
2. View  
3. SuspectEvent  
4. BlockEvent  
5. UnblockEvent  
6. GetStateEvent  
7. StreamingGetStateEvent  
8. SetStateEvent  
9. StreamingSetStateEvent  

Message
View
SuspectEvent
BlockEvent
UnblockEvent
GetStateEvent
StreamingGetStateEvent
SetStateEvent
StreamingSetStateEvent

    同样，也给出一个此方法的应用小例子：

Java代码

1. Object obj;  
2. Message msg;  
3. View v;  
4. obj=channel.receive(0 );  // wait forever   
5. if (obj  instanceof  Message)  
6. msg=(Message)obj;  
7. else   if (obj  instanceof  View)  
8. v=(View)obj;  
9. else   
10. ; // don't handle suspicions or blocks   

Object obj;
Message msg;
View v;
obj=channel.receive(0); // wait forever
if(obj instanceof Message)
msg=(Message)obj;
else if(obj instanceof View)
v=(View)obj;
else
; // don't handle suspicions or blocks

    receive()方法是Channel主动的去取消息，这种方式在现在的JGroups版本中已经不赞成被使用了，而替代方式则是通过setReceiver()方法向Channel注册一个监听器，在有消息到达的时候，自动的调用相应的方法来处理消息。
    setReceiver()方法的参数是一个Receiver接口，此接口继承了MessageListener和 MembershipListener，呵，看名字就知道这两个Listener是做什么的了。JGroups里提供了一个Receiver的 Adapter：ReceiverAdapter，它只是为Receiver接口里的方法提供了一空的实现，可以让我们在自己的实现中只需实现关心的方法 就OK了。下面是一个用注册Receiver的形式接收消息的实例：

Java代码

1. JChannel ch= new  JChannel();  
2. ch.setReceiver(new  ReceiverAdapter() {  
3.     public   void  receive(Message msg) {  
4.         System.out.println("received message "  + msg);  
5.     }  
6.     public   void  viewAccepted(View new_view) {  
7.         System.out.println("received view "  + new_view);  
8.     }  
9.     });  
10. ch.connect("bla" );  

JChannel ch=new JChannel();
ch.setReceiver(new ReceiverAdapter() {
	public void receive(Message msg) {
		System.out.println("received message " + msg);
	}
	public void viewAccepted(View new_view) {
		System.out.println("received view " + new_view);
	}
	});
ch.connect("bla");

    利用上面的的两种方式接收消息时，会将接收到的消息从消息队列中删除，如果只是想了解一下下一条消息，而又不想将它从消息队列中删除时，可以使用peek()方法，它的使用方式同receive()。
    了解了如果发送和接收消息之后，现在Channel想转移到未连接状态啦，可以使用disconnect()，将Channel与组断开连接，这个时候如果再执行发送或接收消息的操作的话，那就等着接收异常吧，呵。
    Channel处于未连接状态之后，可以重新连接到组，也可以通过close()方法关闭Channel，需要注意的是，执行了close()之后，就不 能直接执行connect()方法来使Channel连接到组了，需要使用open()来将Channel重新打开，之后再能与组进行连接。

    上面只是对JGroups的简单应用做一下整理，其实这只是JGroups的皮毛而已，JGroups还包含很多内容：状态传递，Building Blocks(OSCache使用的就是Building Blocks里的NotificationBus)，还有协议栈等等好多内容，这些还得慢慢的继续学习啊。

 

出处：http://www.javaeye.com/topic/81783