Terry.Li-彬

2 API
2．1 Interfaces
2．1．1 Transport
    Transport接口只定义了最简单的方法，用于发送和接收消息。其定义如下：

2．1．2 MessageListener
    如果说Transport接口是以pull的方式接收消息，那么MessageListener则是以push的方式处理消息。当收到消息时，receive方法会被调用。getState() 和setState()方法用于在实例间传递状态。其定义如下：

2．1．3 ExtendedMessageListener
    ExtendedMessageListener继承自MessageListener，它定义了用来在实例间部分传递状态的方法。如果需要传递的状态数据量很大，那么通过配置协议栈，也可以指定使用流的方式传递状态。其定义如下：

2．1．4 MembershipListener
    当收到view、suspicion message和block event 的时候，相应的方法会被调用。这个接口常用的方法是viewAccepted()，以便在新的实例加入（或者离开）到集群时得到通知。当JGroups推 测某个实例可能崩溃时（此时该实例并未离开集群），suspect()方法会被调用，目前没有unsuspect()方法。当JGroups需要通知集群 中的实例不要发送消息时，block()方法会被调用。这通常需要配置FLUSH协议，例如为了确保在进行状态传递的时候，没有实例在发送消息。在 block()方法返回后，所有发送消息的线程都会被阻塞，知道FLUSH协议解除阻塞。需要注意的是，block()方法内不应该执行耗时的操作，否则 整个FLUSH协议都会被阻塞。其定义如下：

2．1．5 ExtendedMembershipListener
    ExtendedMembershipListener继承自MembershipListener。当FLUSH协议解除阻塞的时候，unblock()方法会被调用，所有发送消息的线程可以继续发送消息。其定义如下：

2．1．6 ChannelListener
    可以通过调用JChannel接口的addChannelListener(ChannelListener listener)方法来添加ChannelListener。当Channel被连接或者关闭时，相应的方法会北调用。其定义如下：

2．1．7 Receiver
    Receiver继承自MessageListener和MembershipListener。其定义如下：

2．1．8 ExtendedReceiver
    ExtendedReceiver继承自Receiver、ExtendedMessageListener和ExtendedMembershipListener。其定义如下：

 

2．2 Channel
2．2．1 Creating a channel
    最常见的创建Channel的方法是通过构造函数，此外也可以通过工厂方法。需要注意的是，集群中所有的实例必须有相同的协议栈。JChannel的构造函数之一如下：

    以上的构造函数中，properties参数是冒号分割的字符串，用来配置协议栈。字符串的最左端的元素定义了最底层的协议。如果properties为 null，那么将使用缺省的协议栈，即jgroups-all.jar中的udp.xml。以下是个properties参数的例子：

    此外，也可以用File和URL作为构造函数的参数，这种方式允许以本地或者远程的XML文件配置协议栈。XML文件的config节点中的每个子节点定 义一个协议，第一个子节点定义了最底层的协议。每个子节点名都对应一个Java类名，缺省的协议名不必是全限定类名，它们位于 org.jgroups.stack.protocols包中。如果是自定义的协议，那么则必须是全限定类名。每个协议可以有零个或多个属性，以 name/value对的方式指定。以下是jgroups-all.jar中的udp.xml的内容：

    以上XML文件中，UDP协议的mcast_addr属性被指定使用jgroups.udp.mcast_addr系统属性，如果没有配置这个系统属性，那么使用缺省值228.10.10.10。

2．2．2 Setting options
    通过setOpt(int option, Object value)方法可以给Channel设置属性，目前支持的属性有：

• Channel.BLOCK 这是一个Boolean型的属性，缺省是false。如果设置成true，那么会接收到block message。
• Channel.LOCAL这是一个Boolean型的属性，缺省设置成true。如果是true，那么当集群中的实例向集群发送消息时，这个实例本身也会收到这个消息。
• Channel.AUTO_RECONNECT这是一个Boolean型的属性，缺省是false。如果设置成true，那么shunned channel 在离开集群后，会自动重新加入到集群中。
• Channel.AUTO_GETSTATE 这是一个Boolean型的属性，缺省是false。如果设置成true，那么shunned channel在自动重新加入到集群后，会自动尝试向集群的coordinator 获得集群的状态（需要AUTO_RECONNECT也设置成true）。

    通过Object getOpt(int option)可以或者channel的相关属性值。

2．2．3 Connecting/Disconnecting
    通过调用connect(String cluster_name) throws ChannelException方法连接到集群。cluster_name参数指定了集群的名称。集群中的第一个实例被称为coordinator。当 集群中的成员发生变化的时候，coordinator会向集群中的其它实例发送view message。
    也可以通过调用void connect(String cluster_name, Address target, String state_id,long timeout) throws ChannelException方法连接到集群，并从集群中请求当前的状态。与将这两个操作分开进行相比，在一个方法调用内完成这两个操作，可以允许 JGroups对发送的消息进行优化，更重要的是，从调用者角度来看，这两个操作被合并成一个原子操作。cluster_name参数用于指定集群的名 称。target参数指定了从集群中的哪个实例获得状态，如果该参数为null，那么会从coordinator获得。如果希望传递部分的状态，那么 state_id参数可以用于指定状态id。
    当Channel连接到集群后，通过调用String getClusterName()方法可以获得当前连接到的集群名称。通过调用Address getLocalAddress()方法可以获得channel的地址。通过调用View getView()方法可以获得channel的当前view。每当Channel收到view message的时候，channel的当前view就会被更新。
    通过调用void disconnect()方法以断开到集群的连接。如果channel已经并没有连接到集群，或者chaneel已经被close，那么调用这个方法没有 任何效果。如果channel已经连接到集群，那么这个方法内会向coordinator发送一个离开请求，同时coordinator会向集群中的所有 其它实例发送view message，以通知它们该实例的离开。断开连接的channel可以通过调用connect()方法重新连接到集群。
    通过void close()方法以释放channel占有的资源。Channel被close之后，调用channel上的任何方法都可能会导致异常。

2．2．4 Sending messages
    当channel连接到集群后，可以通过以下方法发送消息。第一个send方法接受一个Message型的参数，如果msg的目标地址不是null，那么 消息会发送到指定地址，否则会发送到集群内的所有成员。msg的源地址可以不必手工设置，如果是null，那么会被协议栈的最底层协议（传输协议）设置成 channel的本地地址。第二个send方法在内部使用了一个send方法。

    以下是个发送消息的例子：

2．2．5 Receiving messages
    Channel 以异步的方式从网络上接收消息，然后把消息存放在一个无界队列中。当调用receive()方法时，会尝试返回队列中取得下一个可用的消息。如果队列中没 有消息，那么会被阻塞。如果timeout小于等于零，那么会永远等待下去；否则会等待timeout指定的毫秒数，直到收到消息或者抛出 TimeoutException。需要注意的是，JChannel. receive(long timeout)方法已经被标记为deprecated。根据channel options的不同，receive()方法可能返回以下类型的对象：

• Message 普通消息。Message.makeReply()可以同于创建消息的应答，即以当前消息的源地址作为应答消息的目标地址。
• View 当集群的成员发生变化的时候，集群的每个成员都会收到view message。当两个或者多个子集群（subgroups）合并成一个的时候，集群中的成员会收到MergeView message。如果需要在子集群合并时处理子集群状态的合并，那么通常需要在单独的线程里执行耗时的操作。
• SuspectEvent 当集群中的某个成员被怀疑崩溃时，集群中的其它成员会收到SuspectEvent message。通过调用SuspectEvent.getMember()可以得到可疑成员的地址。在收到这个消息后，通常还会收到view message。
• BlockEvent 当收到BlockEvent message后，实例应该停止发送消息，然后应该调用Channel.blockOk()方法（目前JChannel.blockOk()方法是一个空 方法）确认已经停止发送消息。当Channel.blockOk()方法调用完毕之后，所有发送消息的线程都会被阻塞直到FLUSH协议解除阻塞。为了接 收BlockEvent message，需要设置Channel.BLOCK属性为true。
• UnblockEvent 当收到UnblockEvent message后，实例可以继续发送消息。
• GetStateEvent 当收到GetStateEvent message后，实例应该保存当前的状态，并将当前状态的一份拷贝作为参数调用Channel.returnState()方法，然后JGroups会 将这个状态返回给请求状态的实例。为了接收GetStateEvent message，需要在协议栈中配置pbcast.STATE_TRANSFER协议。
• StreamingGetStateEvent当收到StreamingGetStateEvent message后，实例应该通过StreamingGetStateEvent.getArg()返回的输出流返回状态。为了接收 StreamingGetStateEvent message，需要在协议栈中配置pbcast.STREAMING_STATE_TRANSFER协议。
• SetStateEvent当收到SetStateEvent message后，实例应该通过SetStateEvent.getArg()返回的字节数组取得状态。
• StreamingSetStateEvent当收到StreamingSetStateEvent message后，实例应该通过StreamingSetStateEvent.getArg()返回的输入流取得状态。为了接收 StreamingSetStateEvent message，需要在协议栈中配置pbcast.STREAMING_STATE_TRANSFER协议。

    以下是个使用pull方式接收消息的例子：

    程序启动后，程序会将在命令行键入键值对（例如key1=value1）保存到HashMap中。并允许在不同的实例间传递状态。
    除了以poll的方式接收消息外，JGroups也支持以push的方式处理消息。通过向JChannel注册Receiver，允许程序以回调的方式处 理消息，而不必启动额外的线程来接收消息，同时JGroups在内部也不用使用无界队列来保存消息。一下是个使用push方式处理消息的例子：

2．2．6 State transfer
    JGroups支持在集群中维护和传递状态(state)，例如web server的Http Sessions等。集群中的某个实例可以通过JChannel.send()方法发送消息，从而把对状态的修改同步到集群的其它实例中。当一个新的实例 加入到集群后，可以调用JChannel.getState()方法向集群中的某个实例（缺省是coordinator）请求获得当前的状态。需要注意的 是，JChannel.getState()方法返回的是boolean类型。如果该实例是集群中的第一个实例，那么该方法返回false（也就是说目前 没有状态），否则返回true。在接下来JChannel.receive()方法的返回值中会包含SetStateEvent message，或者通过MembershipListener.setState()方法获得状态。JChannel.getState()方法不直接 返回状态的原因是，如果JChannel的消息队列中还有未被处理的消息，那么让JChannel.getState()直接返回状态，会破坏消息接收的 FIFO顺序保证，传递的状态也会不正确。

    假设某个集群中包含A、B 和C三个成员，当D加入到集群的时候，如果D调用了JChannel.getState()，那么会发生以下的调用序列：

1. D 调用 JChannel.getState()方法。假设状态从集群中的A成员取得。
2. A 收到GetStateEvent message或者A注册的Receiver的getState() 方法被调用。A返回了当前状态。
3. D 调用 JChannel.getState()方法返回，返回值是true。
4. D 收到SetStateEvent message或者D注册的Receiver的setState()方法被调用。D取得状态。

    2．2．5节的例子中包含了状态传递相关的代码，需要注意的是，在调用JChannel.returnState()方法的时候，为了防止在状态被通过网络发送前，程序的其它地方对状态进行了修改（比如接收到新的消息并更新状态），需要传递当前状态的一份拷贝。
    除了通过处理GetStateEvent 和 SetStateEvent消息来传递状态之外，JGroups也支持通过Reciever传递状态，例如在第一章中演示的例子。
    JGroups支持传递部分状态和以流的形式传递状态，详细内容可以参考JGroups Manual。

3 Building Blocks
    Building blocks位于org.jgroups.blocks包中，在逻辑上可以视为channels之上的一层，它提供了更复杂的接口。Building blocks并不必依赖于channels，部分building blocks只需要实现了Transport接口的类即可工作。以下简要介绍部分building blocks。

3．1 MessageDispatcher
    Channels 通常用于异步地发送和接收消息。然后有些情况下需要同步通信，例如发送者希望向集群发送消息并等待所有成员的应答，或者等待部分成员的应答。 MessageDispatcher支持以同步或者异步的方式发送消息，它在构造时需要一个Channel型的参数。
    MessageDispatcher提供了Object handle(Message msg)方法，用于以push 方式的接收消息并返回应答（必须可以被序列化），该方法抛出的异常也会被传播到消息发送者。MessageDispatcher在内部使用了 PullPushAdapter，PullPushAdapter也是org.jgroups.blocks包中的类，但是已经被标记为 deprecated。这种方式被称为MessageDispatcher的server模式。
    MessageDispatcher的client模式是指通过调用castMessage或者sendMessage向集群发送消息并同步或者异步的等 待应答。castMessage()方法向dests指定的地址发送消息，如果dest为null，那么向集群中所有成员发送消息。 castMessage()方法的返回值是RspList，RspList 实现了Map<Address,Rsp> 接口。msg参数中的目的地址会被覆盖。mode参数（由org.jgroups.blocks.GroupRequest类定义）指定了消息是同步还是 异步发送，其可选值如下：

• GET_FIRST 返回收到的第一个应答。
• GET_ALL 等待所有成员的应答（被怀疑崩溃的成员除外）。
• GET_MAJORITY 等待绝大多数成员（相对与成员的个数）的应答。
• GET_ABS_MAJORITY等待绝大多数成员（一个绝对的数值，只计算一次）的应答。
• GET_N 等待n个应答，如果n大于成员的个数，可能会一直阻塞下去。
• GET_NONE 不等待应答，直接返回，即异步方式。

    castMessage()方法的定义如下：

    sendMessage()方法允许向一个成员发送消息，msg参数的目的地址不能为null。如果mode参数是GET_NONE，那么消息的发送变成 异步方式；否则mode参数会被忽略（缺省采用GET_FIRST）。sendMessage()方法的定义如下：

    以下是个使用MessageDispatcher的例子：

3．2 RpcDispatcher
    RpcDispatcher 继承自MessageDispatcher，它允许远程调用集群中其它成员上的方法，并可选地等待应答。跟MessageDispatcher相比，不需 要为RpcDispatcher指定RequestHandler。RpcDispatcher的构造函数接受一个Object server_obj参数，它是远程调用的目标对象。RpcDispatcher的callRemoteMethods系列方法用于远程调用目标对象上的 方法，该方法可以由MethodCall指定，也可以通过方法名、参数类型指定。跟MessageDispatcher的castMessage()方法 和sendMessage()方法类似，callRemoteMethods系列方法也接受一个int mode参数，其含义也相同。以下是个简单的例子：

3．3 ReplicatedHashMap
    ReplicatedHashMap 继承自ConcurrentHashMap，并在内部使用了RpcDispatcher。ReplicatedHashMap构造函数的 clustername参数指定了集群的名字，集群中所有的实例会包含相同的状态。新加入的实例在开始工作前会从集群中获得当前的状态。对实例的修改（例 如通过put，remove方法）会传播到集群的其它实例中，只读的请求（例如get方法）则是本地调用。需要注意的 是，ReplicatedHashMap的以下划线开头的方法是用于RpcDispatcher的远程调用的。在ReplicatedHashMap上可 以注册 Notification，以便在实例的状态改变时进行回调，所有的回调也是本地的。以下是个简单的例子：

3．4 NotificationBus
    NotificationBus 提供了向集群发送通知的能力，通知可以是任何可以被序列化的对象。NotificationBus在内部使用Channel，其start()和 stop()方法用于启动和停止。NotificationBus的setConsumer()方法用于注册Consumer接口，其定义如下：

    NotificationBus的getCacheFromCoordinator() 和getCacheFromMember()用于请求集群的状态。前者是从coordinator得到状态，后者从指定地址的成员处得到状态。 NotificationBus上注册的Consumer需要实现getCache()方法以返回状态。以下是个简单的例子：

1. import  java.io.BufferedReader;  
2. import  java.io.InputStreamReader;  
3. import  java.io.Serializable;  
4. import  java.util.Iterator;  
5. import  java.util.LinkedList;  
6.   
7. import  org.jgroups.Address;  
8. import  org.jgroups.blocks.NotificationBus;  
9.   
10. public   class  NotificationBusTest  implements  NotificationBus.Consumer {  
11.     //  
12.     private NotificationBus bus;  
13.     private LinkedList<Serializable> cache;  
14.   
15.     public void handleNotification(Serializable n) {  
16.         System.out.println("in handleNotification(" + n + ")");  
17.         if (cache != null) {  
18.             cache.add(n);  
19.         }  
20.     }  
21.   
22.     public Serializable getCache() {  
23.         return cache;  
24.     }  
25.   
26.     public void memberJoined(Address mbr) {  
27.         System.out.println("in memberJoined(" + mbr + ")");  
28.     }  
29.   
30.     public void memberLeft(Address mbr) {  
31.         System.out.println("in memberLeft(" + mbr + ")");  
32.     }  
33.   
34.     @SuppressWarnings("unchecked")  
35.     public void start() throws Exception {  
36.         //  
37.         bus = new NotificationBus("NotificationBusTest", null);  
38.         bus.setConsumer(this);  
39.         bus.start();  
40.         cache = (LinkedList<Serializable>) bus.getCacheFromCoordinator(3000, 1);  
41.         if (cache == null) {  
42.             cache = new LinkedList<Serializable>();  
43.         }  
44.         System.out.println(printCache(cache));  
45.   
46.         //  
47.         sendNotification();  
48.   
49.         //  
50.         bus.stop();  
51.     }  
52.   
53.     private void sendNotification() throws Exception {  
54.         boolean succeed = false;  
55.         BufferedReader br = null;  
56.         try {  
57.             br = new BufferedReader(new InputStreamReader(System.in));  
58.             while (true) {  
59.                 System.out.print("> ");  
60.                 System.out.flush();  
61.                 String line = br.readLine();  
62.                 if (line != null && line.equals("exit")) {  
63.                     break;  
64.                 } else {  
65.                     bus.sendNotification(line);  
66.                 }  
67.             }  
68.             succeed = true;  
69.         } finally {  
70.             if (br != null) {  
71.                 try {  
72.                     br.close();  
73.                 } catch (Exception e) {  
74.                     if (succeed) {  
75.                         throw e;  
76.                     }  
77.                 }  
78.             }  
79.         }  
80.     }  
81.   
82.     private String printCache(LinkedList<Serializable> c) {  
83.         StringBuffer sb = new StringBuffer();  
84.         sb.append("[");  
85.         for (Iterator<Serializable> iter = c.iterator(); iter.hasNext();) {  
86.             sb.append(iter.next());  
87.             if (iter.hasNext()) {  
88.                 sb.append(",");  
89.             }  
90.         }  
91.         sb.append("]");  
92.         return sb.toString();  
93.     }  
94.   
95.     public static void main(String[] args) {  
96.         try {  
97.             new NotificationBusTest().start();  
98.         } catch (Exception e) {  
99.             e.printStackTrace();  
100.         }  
101.     }  
102. }