欢迎您光临本小站。希望您在这里可以找到自己想要的信息。。。

JGroup初步学习

架构&设计模式 water 3713℃ 0评论

现在有很多项目都使用JGroups做底层的通讯,知道的开源项目有JBoss
Cache和OSCache
用它做为底层支持来实现集群的,一定还有其他的项目也用到了,只不过我不知道而已了。


JGroups 适合使用场合
服务器集群cluster、多服务器通讯、服务器replication(复制)等,分布式cache缓存

JGroups 简介
JGroups是一个基于Java语言的提供可靠多播(组播)的开发工具包。在IP
Multicast基础上提供可靠服务,也可以构建在TCP或者WAN上。主要是由Bela
Ban开发,属于JBoss.org,在JBoss的网站也有一些相关文档。目前在
SourceForge上还是比较活跃,经常保持更新。



JGroups是一个可靠的组间通讯工具,进程可以加入一个通讯组,给组内所有的成员或单独的成员发送消息,同样,也可以从组中的成员处接收消息。系统会记录组的每一个成员,在新成员加入或是现有的成员离开或是崩溃时,会通知组内的其他成员,这样我们就不必自己去管理这些事情了。

要想加入一个组,并与组内其他的成员交互,必须建立一个Channel连接到组,同一个组内的所有成员使用相同的组名称。首先是创建一个Channel,可以直接实例化一个Channel的实现,这里用的是JChannel:

Java代码  收藏代码

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


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

Java代码  收藏代码

  1. 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;  



两个方法基本是一样的,只不过一个是直接提供一个消息,而另一个只是提供了消息的目的地,源,和消息内容,其实这个方法在内部也是通过第一个方法来实现的,在其内部,将提供的这三个参数组合成一个消息,再调用第一个方法,具体使用哪个方法,则看个人喜好和实际情况了。

消息的由消息的目的地,源,Flag,消息内容,Header组成。其中如果目的地为空,则认为是发给所有组成员的消息;源为空的话,在底层的协议将其放到网络上时,会自动的将本Channel的地址填充进去。Address则是组成员的地址,用于唯一的标识一个组成员的接口,JGroups提供了几种默认的实现。下面的例子是发送一条消息到组内所有的成员处: 

Java代码  收藏代码

  1. Hashtable data; // any serializable data  

  2. try  

  3.     channel.send(nullnulldata);  

  4.  

  5. catch(Exception ex)  

  6. // handle errors  



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

Java代码  收藏代码

  1. Address receiver;  

  2. Hashtable data;  

  3. try  

  4. receiver=channel.getView().getMembers().first();  

  5. channel.send(receiver, nulldata);  

  6.  

  7. catch(Exception ex)  

  8. // handle errors  

  9.  



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

可以发送消息,同样也可以接收消息: 

Java代码  收藏代码

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



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

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  



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");  



利用上面的的两种方式接收消息时,会将接收到的消息从消息队列中删除,如果只是想了解一下下一条消息,而又不想将它从消息队列中删除时,可以使用peek()方法,它的使用方式同receive()。
 

了解了如果发送和接收消息之后,现在Channel想转移到未连接状态啦,可以使用disconnect(),将Channel与组断开连接,这个时候如果再执行发送或接收消息的操作的话,那就等着接收异常吧,呵。

Channel处于未连接状态之后,可以重新连接到组,也可以通过close()方法关闭Channel,需要注意的是,执行了close()之后,就不
能直接执行connect()方法来使Channel连接到组了,需要使用open()来将Channel重新打开,之后再能与组进行连接。 


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

转自:http://www.iteye.com/topic/81783


JGroups使用例子, JGroups demo, Tim的hello world例子
Timreceiver.java

import org.jgroups.tests.perf.Receiver;
import org.jgroups.tests.perf.Transport;
import org.jgroups.util.Util;

public class TimReceiver implements Receiver {

      private Transport transport = null;

      

      public static void main(String[] args) {

            TimReceiver t = new TimReceiver();

            try {

                  int sendMsgCount = 5000;

                  int msgSize = 1000;

                  t.start();


                  t.sendMessages(sendMsgCount, msgSize);

                  System.out.println("########## Begin to recv...");

                  Thread.currentThread().join();

            } catch (Exception e) {

                  e.printStackTrace();

            } finally {

                  if (t != null) {

                        t.stop();

                  }

            }

      }

      

      public void start()

                  throws Exception {

            transport = (Transport) new TimTransport();

            transport.create(null);

            transport.setReceiver(this);

            transport.start();

      }


      public void stop() {

            if (transport != null) {

                  transport.stop();

                  transport.destroy();

            }

      }


      private int count = 0;

      public void receive(Object sender, byte[] data) {

            System.out.print(".");

            if (++count == 5000) {

                  System.out.println("\r\nRECV DONE.");

                  System.exit(0);

            }

            

      }


      private void sendMessages(int count, int msgSize)

                  throws Exception {

            byte[] buf = new byte[msgSize];

            for (int k = 0; k < msgSize; k++)

                  buf[k] = 'T';


            System.out.println("-- sending " + count + " " + Util.printBytes(msgSize) + " messages");


            for (int i = 0; i < count; i++) {

                  transport.send(null, buf);

            }

            

            System.out.println("######### send complete");

      }
}


TimTransport.java


import java.util.Map;
import java.util.Properties;

import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.tests.perf.Receiver;
import org.jgroups.tests.perf.Transport;

public class TimTransport extends ReceiverAdapter implements Transport{

      private JChannel channel = null;

      private String groupName = "TimDemo";

      private Receiver receiver = null;

      

      String PROTOCOL_STACK_UDP1 = "UDP(bind_addr=192.168.100.59";  

      String PROTOCOL_STACK_UDP2 = ";mcast_port=8888";

      String PROTOCOL_STACK_UDP3 = ";mcast_addr=225.1.1.1";

      String PROTOCOL_STACK_UDP4 = ";tos=8;loopback=false;max_bundle_size=64000;" +

                  "use_incoming_packet_handler=true;use_outgoing_packet_handler=false;ip_ttl=2;enable_bundling=true):"

           + "PING:MERGE2:FD_SOCK:FD:VERIFY_SUSPECT:"

             +"pbcast.NAKACK(gc_lag=50;max_xmit_size=50000;use_mcast_xmit=false;" +

                         "retransmit_timeout=300,600,1200,2400,4800;discard_delivered_msgs=true):"

             +"UNICAST:pbcast.STABLE:VIEW_SYNC:"

             +"pbcast.GMS(print_local_addr=false;join_timeout=3000;" +

                         "join_retry_timeout=2000;" +

                         "shun=true;view_bundling=true):"

             +"FC(max_credits=2000000;min_threshold=0.10):FRAG2(frag_size=50000)";


            

      public Object getLocalAddress() {

            return channel != null ? channel.getLocalAddress() : null;

      }


      public void start() throws Exception {

            channel.connect(groupName);            

      }


      public void stop() {

            if (channel != null) {

                  channel.shutdown();

            }

      }


      public void destroy() {

            if (channel != null) {

                  channel.close();

                  channel = null;

            }

      }


      public void setReceiver(Receiver r) {

            this.receiver = r;

      }


      public Map dumpStats() {

            return channel != null ? channel.dumpStats() : null;

      }


      public void send(Object destination, byte[] payload) throws Exception {

            byte[] tmp = new byte[payload.length];

            System.arraycopy(payload, 0, tmp, 0, payload.length);

            Message msg = null;

            msg = new Message((Address) destination, null, tmp);

            if (channel != null) {

                  channel.send(msg);

            }

      }


      public void receive(Message msg) {

            Address sender = msg.getSrc();

            byte[] payload = msg.getBuffer();

            if (receiver != null) {

                  try {

                        receiver.receive(sender, payload);

                  } catch (Throwable tt) {

                        tt.printStackTrace();

                  }

            }

      }


      public void create(Properties config) throws Exception {

            String PROTOCOL_STACK = PROTOCOL_STACK_UDP1 + PROTOCOL_STACK_UDP2 + PROTOCOL_STACK_UDP3 + PROTOCOL_STACK_UDP4;

            channel = new JChannel(PROTOCOL_STACK);

            channel.setReceiver(this);            

      }


      public void send(Object destination, byte[] payload, boolean oob) throws Exception {

            send(destination, payload);

      }
}

转自

转载请注明:学时网 » JGroup初步学习

喜欢 (0)or分享 (0)

您必须 登录 才能发表评论!