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. }