我们在开发集群系统的过程中，往往需要在多个进程间同步很多状态，比如每个服务器进程的负载状况、数据队列长度等等……。还有一些情况，我们需要把不同的进程分类，然后分发不同的通知消息，最常见的是发出一些运维命令，如回写数据、清理缓存……。在游戏服务集群中，需要群发消息的功能更是常见，比如全副广播、玩家在线列表维护……。因此我们往往需要实现“分组”“广播”的功能。在开源软件库中，有一款专门完成这类工作的产品，叫做JGroups。
这是一款Apache基金会下的Java语言开源库，官网地址是：http://jgroups.org/
集群服务中，各节点之间的通信能力至关重要。如果个个节点（进程）无法通信，集群将无法协同工作。——这正是JGroups的目标。在一般的情况下，我们可能会选择使用“消息队列”，比如JAVA就有JMS的消息队列规范，而Apache的开源软件ActiveMQ正是实现JMS规范的著名软件。但是，消息队列在使用的时候会显得太“重”，他需要额外启动的队列进程：存储队列消息，特定的进程间通信接口。对于希望能构建轻量级集群的方案里，部署一堆消息队列进程，然后配置管理他们，都是非常麻烦的。而jgroups则正好填补了这个空缺：它是一个库，使用API接口就好了，无须对协议编程；它让你的业务进程直接有通信的能力。
jgroups本身的通信能力，是基于UDP的，因此也支持使用UDP组播能力。这在轻量的服务集群中非常实用：往往一批服务器都连在同一个交换机上。这种情况下的UDP组播丢包率低，时延也低，而且能有效利用内网带宽。
jgroups自身的架构，和很多开源的通信库一样，是分层的。应用程序使用JChannel构建的使用界面API，以“频道”的模型来使用通信能力。底层则分为多个不同能力的实现层。
组播功能
我们先来看看如何用jgroups实现通信功能。jgroups是一个库，因此完全通过API编程就能让你的进程实现通信能力，无须安装部署任何其他软件。
首先，你需要建立一个频道，并连接进去。这个频道无须额外配置，只要调用代码就可以了。其次，构造一条需要发送的消息，这个消息包括内容、发送方地址、接收方地址。接收方地址标识在对应的频道中的具体接收目标，如果不填就是广播，频道中所有节点都会收到，消息的内容就是byte[]，也就是任何的数据类型都可以。最后，注册一个接受消息的接口回调对象，让这个对象也连接到同样名字的频道就可以了。
//加入集群
JChannelchannel = new JChannel();
channel.connect(“pushserver2”);
//发送消息，参数一是接收方地址，参数二是发送方地址；如果接收方地址是null，表示广播
Message msg= new Message(null, channel.getAddress(), “hello”.getBytes());
channel.send(msg);
//接收消息，Receiver类是实现了ReceiverAdapter接口的
ReceiverAdapter receiver = new Receiver();
channel.setReceiver(receiver);
channel.connect(“pushserver2”);
//接收回调函数
@Override
public void receive(Message msg)
{
byte[]buf = msg.getBuffer();
}
在以上操作的过程中，每个节点（进程）对于“频道”的状态，是在不断变化的。下图表述了这种变化：除了直接用JChannel来收发消息外，jgroups还提供了一种叫BuildingBlocks的使用方法。这个其实是对JChannel更高级的一种封装。主要分为Message Dispatcher和Rpc Dispatcher。
MessageDispatcher可以直接不显式建频道进行广播和单播，每次调用就完成一次简单的调用。这样就无须编码维护JChannel的状态。RpcDispatcher则更好用，他通过JAVA的反射功能，直接调用目标节点上的方法。他们的使用接口如下：
MessageDispatcher
public <T>RspList<T> castMessage(final Colllection<Address>dests, Message msg, RequestOptions options) //广播
public <T>T sendMessage(Message msg, RequestOptions opts) //单播
RpcDispatcher
public <T> RspList<T> callRemoteMethods(
Collection<Address> dests,
String method_name,
Object[] args,
Class[] types,
RequestOptionsoptions) throws Exception;
发现功能
在集群系统中，我们除了要给节点发消息，还需要随时了解集群中节点的状况。因此jgourps也提供了对应的接口，让我们了解这些情况。用法就是实现MembershipListener这个接口，其中的回调方法就会由jroups在发生事件时调用。
//状态回调接口
public interface MembershipListener
{
void viewAccepted(View new_view); //集群成员有变化
void suspect(Object suspected_mbr); //可能有节点挂了
void block(); //表示发送消息将被阻塞
void unblock(); //表示发送消息将不再被阻塞
}状态传递
在集群系统里面，有时候我们不仅仅是通过消息来传递消息，还需要“同步”一些状态。也就是说不是说发一个内容给对方，而是让多个节点获得一致的数据，这个状态是在同一个“组”里面共享的。如果用广播API去做这个事情，自己还需要编写一堆代码。所以jgroups直接提供了一套易用的API来完成这个事。
首先，任何一个节点都可以对任何一个组，发起状态同步的请求。然后，在组中的“最老”的节点，就会有一个回调函数被调用，这个回调函数负责发送状态数据给请求者。状态数据可以是任何类型，是通过一个OutputStream来发送的。当然我们常常会使用JDK的ObjectOutputStream来直接发送一个状态对象。最后，这个状态请求者就会收到一个回调，数据由参数的InputStream传入。这套API的特点是，发送数据和接收数据者并不需要互相知道，而是通过一个组关联起来，这样就避免了复杂的地址管理工作。因为组中的成员可能随时退出和加入，要维持一个公共的数据变得很麻烦。
1. 发起状态请求：
public void getState(Addresstarget, long timeout) throws Exception;
2. 触发某一节点（最老）回调：
public void getState(OutputStreamoutput) throws Exception;
3. 触发请求者的回调：
public void setState(InputStreaminput) throws Exception;
协议
由于jgroups是一个组播的通信库，因此其协议栈包含了其内部支持的功能。这里列举一下：
消息传输：UDP/TCP/TUNNEL
PING ：发现协议，IP多播
MERGE3：合并子集群回归一整个集群
FD_SOC：故障检测
VERIFY_SUSPECT：Double-checks故障节点
BARRIER：状态传输
UFC/MFC：单播、多播流量控制协议
可以看到，为了实现组播，其实底层是需要有很多网络维护工作的，特别是集群中节点故障状态的控制，由于jgroups帮我们做了，所以我们才可以这么简单的完成可靠性如此之高的组播功能。
思考
jgrops的功能固然强大，但是让我更感兴趣的是其设计思想，这个产品体现出很多设计上的闪光点：
1. 如何简化集群管理接口。在很多方案中，集群管理库都是使用起来非常复杂的，但是jgroups设计了几个模型，大大简化了使用者需要理解的概念。
a) 使用IP组播或dir地址标识集群
b) 多个功能的集群可以合并到一个集群做管理，区分“组”单位即可
c) 自定义字符串节点组的名字
d) 通知集群变化的回调
e) 基于byte[]和对象的操作API
2. 对于集群环境的启示。我们在实际工作中，集群往往是按需搭建的，因此缺乏统一、有效的集群模型，其中有技术上实现难度的原因，也有思维上缺乏构建模型的原因，但是jgroups让我们发现，集群的管理功能，如状态同步、消息传递、底层通信都可以模块化设计的。
a) 可重用的集群状态管理和通信组件库
b) 隔离底层具体技术实现，而以Channel为单位管理
c) 丰富的非功能需求集成
前言
JGroups是一个开源的纯java编写的可靠的群组通讯工具。其工作模式基于IP多播，但可以在可靠性和群组成员管理上进行扩展。其结构上设计灵活，提供了一种灵活兼容多种协议的协议栈。
JGroups 多线程的方式实现了多个协议之间的协同工作，常见工作线程有心跳检测，诊断等等。
JGroups实现多机器之间的通信一般都会包含维护群组状态、群组通信协议、群组数据可靠性传输这样的一些主题。
JGroups群组的各个节点是存在"管理节点"的，至少可以说某个节点提供了在一段时间内维护状态信息和消息可靠性检测的功能(一般是最先启动的节点)。
目前Jboss、Ecache的分布式缓存是基于Groups通信。
若JGroups通信基于Udp，则可能需要开启机器上UDP相关的设置，比如Open udp。
温馨提示：JGroups各个协议相关的配置文件都可以从JGroups-x.x.x.Final.jar中找到。
JGroups 资料
http://www.jgroups.org/tutorial/index.html(官网)
http://sourceforge.net/projects/javagroups/(JGroups工程&讨论组(Discussion))
JGroups 入门示例
1,节点通信(tcp/ip,udp)方式.
2,通道和消息传送.
3,节点状态同步.
tcp/ip与udp协议
通常我们都知道tcp和udp最大的区别在于可靠性，tcp是基于可靠连接的传输，udp则属非连接，具体可参考百度百科(http://baike.baidu.com/view/1161229.htm?fr=aladdin)。
JGroups当中，udp是比较推荐的通信方式，其特点是不需要知道另一个节点的ip，通过多播网络发现就可以“找到”相应的节点，而tcp则需要在配置文件中固定配置。
示例代码(之后的测试基于tcp,因为不同机器的测试由于udp端口的问题未成功)
tcp配置文件network-tcp.xml
<!--
TCP based stack, with flow control and message bundling. This is usually used when IP
multicasting cannot be used in a network, e.g. because it is disabled (routers discard multicast).
Note that TCP.bind_addr and TCPPING.initial_hosts should be set, possibly via system properties, e.g.
-Djgroups.bind_addr=192.168.5.2 and -Djgroups.tcpping.initial_hosts=192.168.5.2[7800]
author: Bela Ban
--><config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-3.3.xsd"><TCP bind_addr="192.168.19.112"
bind_port="7800"
loopback="false"
recv_buf_size="${tcp.recv_buf_size:5M}"
send_buf_size="${tcp.send_buf_size:640K}"
max_bundle_size="64K"
max_bundle_timeout="30"
use_send_queues="true"
sock_conn_timeout="300"
timer_type="new3"
timer.min_threads="4"
timer.max_threads="10"
timer.keep_alive_time="3000"
timer.queue_max_size="500"
thread_pool.enabled="true"
thread_pool.min_threads="1"
thread_pool.max_threads="10"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="false"
thread_pool.queue_max_size="100"
thread_pool.rejection_policy="discard"
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="discard"/><TCPPING timeout="3000"
initial_hosts="${jgroups.tcpping.initial_hosts:192.168.19.112[7800],192.168.19.112[7801]}"
port_range="1"
num_initial_members="10"/><MERGE2 min_interval="10000"
max_interval="30000"/><FD_SOCK/><FD timeout="3000" max_tries="3"/><VERIFY_SUSPECT timeout="1500"/><BARRIER /><pbcast.NAKACK2 use_mcast_xmit="false"
discard_delivered_msgs="true"/><UNICAST3 /><pbcast.STABLE stability_delay="1000" desired_avg_gossip="50000"
max_bytes="4M"/><pbcast.GMS print_local_addr="true" join_timeout="3000"
view_bundling="true"/><MFC max_credits="2M"
min_threshold="0.4"/><FRAG2 frag_size="60K"/><!--RSVP resend_interval="2000" timeout="10000"/--><pbcast.STATE_TRANSFER/></config>
udp配置文件network-udp.xml
<config xmlns="urn:org:jgroups"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="urn:org:jgroups http://www.jgroups.org/schema/JGroups-3.5.xsd"><UDP
mcast_addr="${jgroups.udp.mcast_addr:235.5.5.5}"
mcast_port="${jgroups.udp.mcast_port:45588}"
tos="8"
ucast_recv_buf_size="20M"
ucast_send_buf_size="640K"
mcast_recv_buf_size="25M"
mcast_send_buf_size="640K"
loopback="true"
max_bundle_size="64K"
max_bundle_timeout="30"
ip_ttl="${jgroups.udp.ip_ttl:2}"
enable_diagnostics="true"
thread_naming_pattern="cl"
timer_type="new"
timer.min_threads="4"
timer.max_threads="10"
timer.keep_alive_time="3000"
timer.queue_max_size="500"
thread_pool.enabled="true"
thread_pool.min_threads="2"
thread_pool.max_threads="8"
thread_pool.keep_alive_time="5000"
thread_pool.queue_enabled="true"
thread_pool.queue_max_size="10000"
thread_pool.rejection_policy="discard"
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_ALL/><VERIFY_SUSPECT timeout="1500"/><BARRIER /><pbcast.NAKACK use_mcast_xmit="true"
retransmit_timeout="300,600,1200"
discard_delivered_msgs="true"/><pbcast.STABLE stability_delay="1000"
desired_avg_gossip="50000"
max_bytes="4M"/><pbcast.GMS print_local_addr="true"
print_physical_addrs="true"
join_timeout="3000"
view_bundling="true"
max_join_attempts="3"/><UFC max_credits="2M" min_threshold="0.4"/><MFC max_credits="2M" min_threshold="0.4"/><FRAG2 frag_size="60K"/><pbcast.STATE_TRANSFER /></config>
数据节点Node.java
package org.wit.ff;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.ReentrantLock;
import org.apache.log4j.Logger;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.ReceiverAdapter;
import org.jgroups.View;
import org.jgroups.util.Util;
/**
*
* <pre>
* 节点.
* </pre>
*
* @author F.Fang
* @version $Id: CacheNode.java, v 0.1 2014年10月17日 上午5:27:11 F.Fang Exp $
*/
public class Node extends ReceiverAdapter {
private final static Logger LOG = Logger.getLogger(Node.class);
/**
* 配置文件.
*/
private static final String CONFIG_XML = "network-tcp.xml";
/**
* 集群名称.
*/
private static final String CLUSTER_NAME = "FF";
/**
* 节点通道.
*/private JChannel channel = null;
/**
* 以此作为节点间初始化的同步数据.
*/
private Map<String, String> cacheData = new HashMap<String, String>();
private ReentrantLock lock = new ReentrantLock();
public Node() {
InputStream is = this.getClass().getClassLoader().getResourceAsStream(CONFIG_XML);
try {
channel = new JChannel(is);
channel.setReceiver(this);
channel.connect(CLUSTER_NAME);
channel.getState(null,50000);
} catch (Exception e) {
LOG.error("启动节点异常!", e);
// 最好是自定义RuntimeException!thrownew RuntimeException("启动节点异常!", e);
}
}
/**
*
* <pre>
* 发送消息给目标地址.
* </pre>
*
* @param dest
* 为空表示发给所有节点.
* @param textMsg
* 消息.
*/
public void sendMsg(Address dest, Object textMsg) {
Message msg = new Message(dest, null, textMsg);
try {
channel.send(msg);
} catch (Exception e) {
LOG.error("消息发送失败!", e);
// 应自定异常,最好是自定义Exception类型!thrownew RuntimeException("消息发送失败!", e);
}
}
@Override
public void getState(OutputStream output) throws Exception {
//cacheData过大可能会造成节点的状态同步时间过长.
lock.lock();
try {
Util.objectToStream(state, new DataOutputStream(output));
}catch(Exception e){
throw e;
}finally{
lock.unlock();
}
}
@Override
public void receive(Message msg) {
//当前节点不接收自己发送到通道当中的消息.if (msg.getSrc().equals(channel.getAddress())) {
return;
}
LOG.info(msg.getObject());
}
@Override
public void setState(InputStream input) throws Exception {
lock.lock();
try {
@SuppressWarnings("unchecked")
Map<String, String> cacheData = (Map<String, String>) Util.objectFromStream(new DataInputStream(input));
this.cacheData.putAll(cacheData);
} catch (Exception e) {
LOG.error("从主节点同步状态到当前节点发生异常!", e);
} finally {
lock.unlock();
}
}
@Override
public void viewAccepted(View view) {
LOG.info("当前成员[" + this.channel.getAddressAsString() + "]");
LOG.info(view.getCreator());
LOG.info(view.getMembers());
LOG.info("当前节点数据:" + cacheData);
}
/**
*
* <pre>
* 提供一个简单的初始化数据的方法.
* </pre>
*
*/
public void addData(String key,String val){
if(key!=null&&!key.isEmpty()){
cacheData.put(key, val);
}
}
}
实例节点1 Node1.java
package org.wit.ff;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
import org.wit.ff.Node;
/**
*
* <pre>
* tcp模式下:
* 如果是同一台机器测试,请注意在
* TCPPING 元素下修改 initial_hosts的配置端口:
* 例如:"${jgroups.tcpping.initial_hosts:192.168.19.100[7800],192.168.19.100[7801]}
* 如果是多台机器测试,请注意在
* TCPPING 元素下修改 initial_hosts的ip,端口随意:
* 例如:"${jgroups.tcpping.initial_hosts:192.168.19.100[7800],192.168.19.178[7800]}
*
* udp模式下:
* 同一台机器的不同端口(端口是动态的)可通信.
* 不同机器之间的ip多播可能会受到一些因素限制而造成节点之间无法彼此发现.
* </pre>
*
* @author F.Fang
* @version $Id: Node1.java, v 0.1 2014年10月15日 上午5:31:32 F.Fang Exp $
*/
public class Node1 {
public static void main(String[] args) {
Node node = new Node();
node.addData("hello", "world");
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 使用控制台发送消息给Node2.
Scanner scanner = new Scanner(System.in);
while(true){
String text = scanner.next();
if("exit".equals(text)){
break;
}
node.sendMsg(null,"hello "+text+",node2!");
}
}
}
实例节点2 Node2.java
package org.wit.ff;
import java.util.Scanner;
import java.util.concurrent.TimeUnit;
/**
*
* <pre>
* tcp模式下:
* 如果是同一台机器测试,请注意在
* TCPPING 元素下修改 initial_hosts的配置端口:
* 例如:"${jgroups.tcpping.initial_hosts:192.168.19.100[7800],192.168.19.100[7801]}
* 如果是多台机器测试,请注意在
* TCPPING 元素下修改 initial_hosts的ip,端口随意:
* 例如:"${jgroups.tcpping.initial_hosts:192.168.19.100[7800],192.168.19.178[7800]}
*
* @author F.Fang
* @version $Id: Node2.java, v 0.1 2014年10月15日 上午5:31:44 F.Fang Exp $
*/
public class Node2 {
public static void main(String[] args) {
Node node = new Node();
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 使用控制台发送消息给Node1.
Scanner scanner = new Scanner(System.in);
while (true) {
String text = scanner.next();
if ("exit".equals(text)) {
break;
}
node.sendMsg(null,"hello " + text + ",node1!");
}
}
}
测试Case
启动Node1，Node1平稳后启动Node2。
Node1运行信息如下：
DEBUG Configurator - set property TCP.diagnostics_addr to default value /ff0e:0:0:0:0:0:75:75
-------------------------------------------------------------------
GMS: address=DSH07fFang-18185, cluster=FF, physical address=192.168.19.112:7800
-------------------------------------------------------------------
DEBUG NAKACK2 -
[DSH07fFang-18185 setDigest()]
existing digest: []
new digest: DSH07fFang-18185: [0 (0)]
resulting digest: DSH07fFang-18185: [0 (0)]
DEBUG GMS - DSH07fFang-18185: installing view [DSH07fFang-18185|0] (1) [DSH07fFang-18185]
DEBUG STABLE - resuming message garbage collection
DEBUG FD_SOCK - VIEW_CHANGE received: [DSH07fFang-18185]
INFO Node - 当前成员[DSH07fFang-18185]
INFO Node - DSH07fFang-18185
INFO Node - [DSH07fFang-18185]
INFO Node - 当前节点数据:{}
DEBUG STABLE - resuming message garbage collection
DEBUG GMS - created cluster (first member). My view is [DSH07fFang-18185|0], impl is org.jgroups.protocols.pbcast.CoordGmsImpl
DEBUG STABLE - suspending message garbage collection
DEBUG STABLE - DSH07fFang-18185: resume task started, max_suspend_time=33000
DEBUG GMS - DSH07fFang-18185: installing view [DSH07fFang-18185|1] (2) [DSH07fFang-18185, DSH07fFang-2882]
DEBUG FD_SOCK - VIEW_CHANGE received: [DSH07fFang-18185, DSH07fFang-2882]
INFO Node - 当前成员[DSH07fFang-18185]
INFO Node - DSH07fFang-18185
INFO Node - [DSH07fFang-18185, DSH07fFang-2882]
INFO Node - 当前节点数据:{hello=world}
DEBUG FD_SOCK - ping_dest is DSH07fFang-2882, pingable_mbrs=[DSH07fFang-18185, DSH07fFang-2882]
DEBUG STABLE - resuming message garbage collection
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
DEBUG FD - DSH07fFang-18185: sending are-you-alive msg to DSH07fFang-2882
主要包括ip通信信息、状态、心跳等等。Node2运行消息如下：
DEBUG Configurator - set property TCP.diagnostics_addr to default value /ff0e:0:0:0:0:0:75:75
-------------------------------------------------------------------
GMS: address=DSH07fFang-2882, cluster=FF, physical address=192.168.19.112:7801
-------------------------------------------------------------------
DEBUG GMS - DSH07fFang-2882: sending JOIN(DSH07fFang-2882) to DSH07fFang-18185
DEBUG NAKACK2 -
[DSH07fFang-2882 setDigest()]
existing digest: []
new digest: DSH07fFang-18185: [0 (0)], DSH07fFang-2882: [0 (0)]
resulting digest: DSH07fFang-18185: [0 (0)], DSH07fFang-2882: [0 (0)]
DEBUG GMS - DSH07fFang-2882: installing view [DSH07fFang-18185|1] (2) [DSH07fFang-18185, DSH07fFang-2882]
DEBUG FD_SOCK - VIEW_CHANGE received: [DSH07fFang-18185, DSH07fFang-2882]
INFO Node - 当前成员[DSH07fFang-2882]
INFO Node - DSH07fFang-18185
INFO Node - [DSH07fFang-18185, DSH07fFang-2882]
INFO Node - 当前节点数据:{hello=world}
DEBUG FD_SOCK - ping_dest is DSH07fFang-18185, pingable_mbrs=[DSH07fFang-18185, DSH07fFang-2882] DEBUG FD - DSH07fFang-2882: sending are-you-alive msg to DSH07fFang-18185 DEBUG FD - DSH07fFang-2882: sending are-you-alive msg to DSH07fFang-18185节点之间存在通信和状态同步，可以通过控制台输入发送消息的命令观察节点变化。
