应用jgroups实现tomcat session的同步

项目中采用将会话保存到全局缓存中的方式，实现了分布式session的功能。由于整个系统划分为多个子系统，并且分别部署到不同的服务器节点上，所以各tomcat容器之间的session的注销和过期时间需要同步。目前项目中采用了jgroups的组播方式实现了各容器之间session的同步问题，下面是部分代码，仅供参考：

 

import org.apache.catalina.Manager;
import org.apache.catalina.Session;
import org.apache.catalina.session.CacheConstants;
import org.apache.catalina.session.CacheSession;
import org.apache.juli.logging.Log;
import org.apache.juli.logging.LogFactory;
import org.jgroups.ChannelClosedException;
import org.jgroups.ChannelException;
import org.jgroups.ChannelNotConnectedException;
import org.jgroups.ExtendedReceiverAdapter;
import org.jgroups.JChannel;
import org.jgroups.Message;
import org.jgroups.View;
import org.jgroups.util.Util;


public class CacheMsgManager extends ExtendedReceiverAdapter{
	protected Log logger = LogFactory.getLog(CacheMsgManager.class);
	
	private static Manager manager = null;
	private static CacheMsgManager msgManager = null;
	private static String PROTOCOL_STACK = "";
	private static final String SESSION_GROUP = "sessionid-jgroup";
	private static JChannel channel;
	private List<String> msgList = new ArrayList<String>();//模拟状态对象，保存的是本节点的收到的消息
	
	
	public static CacheMsgManager getInstance(String ip){
		if(msgManager == null)
			msgManager = new CacheMsgManager(ip);
		
		return msgManager;
	}
	
	private CacheMsgManager(String ip){
		PROTOCOL_STACK = "UDP(mcast_addr="+ ip +";mcast_port=32767;ip_ttl=3;"+
			"mcast_send_buf_size=150000;mcast_recv_buf_size=80000):"+
			"PING(timeout=2000;num_initial_members=3):"+
			"MERGE2(min_interval=5000;max_interval=10000):"+
			"FD_SOCK:"+
			"VERIFY_SUSPECT(timeout=1500):"+
			"pbcast.NAKACK(gc_lag=50;retransmit_timeout=300,600,1200,2400,4800):"+
			"pbcast.STABLE(desired_avg_gossip=20000):"+
			"UNICAST(timeout=2500,5000):"+
			"FRAG:"+
			"pbcast.GMS(join_timeout=5000;join_retry_timeout=2000;shun=false;print_local_addr=false):"+
			"pbcast.STATE_TRANSFER";
		
		genChannel();
	}
	
	public static JChannel getChannel() {
		return channel;
	}
	
	
	/**
	 * 在有节点向本节点请求状态的时候被调用
	 */
	@Override
	public byte[] getState() {
		byte[] state = null;
		synchronized(msgList){
			try {
				state = Util.objectToByteBuffer(msgList);
			} catch (Exception e) {
				e.printStackTrace();
			}
			return state;
		}
	}
	
	/**
	 * 在其他节点返回状态给本节点的时候被调用
	 */
	@Override
	public void setState(byte[] state) {
		synchronized(msgList){
			try {
				List<String> tmpList = (List<String>)Util.objectFromByteBuffer(state);
				msgList.clear();
				msgList.addAll(tmpList);   
				logger.info("^^^^^^^^^^^^^^^^^^^ receive state:[");
				
				for(String msg : msgList){
					logger.info(msg);
				}
				logger.info("]");
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 当有消息进来的时候被调用  
	 */  
	@Override
	public void receive(Message msg) {
		logger.info("^^^^^^^^^^^^^^^^^^^ new message receive from "+ msg.getSrc() +" : "+ msg.getObject());
		//msgList.add(msg.getSrc()+" send : "+ msg.getObject());
		
		String msgBody = "";
		String oper = "";
		
		if(msg != null){
			msgBody = (String)msg.getObject();
			String[] body = msgBody.split(",");
			
			if(body != null && body.length > 0){
				oper = body[0];
				
				String sessionId = null;
				String[] ids = null;
				
				if(body[1] != null){
					ids = body[1].split(":");
					if(ids != null)
						sessionId = ids[1];
				}
				
				logger.info("^^^^^^^^^^^^^^^^^^^ receive() sessionId : "+ sessionId);
				if(oper != null && oper.length() > 0){				
					
					if(oper.equals(CacheConstants.OPER_KEY_EXPIRE)){
						operExpire(sessionId);
					}
					
					if(oper.equals(CacheConstants.OPER_KEY_ACCESS)){
						operAccess(sessionId);
					}
				}
			}			
		}
	}
	
	
	private void genChannel(){
		try{
			if(channel == null){
				channel = new JChannel(PROTOCOL_STACK);//打开channel并指定配置文件
				channel.connect(SESSION_GROUP);//指定组名，就可以创建或连接到广播组
				channel.setReceiver(this);//注册回调接口，使用"推"模式来接受广播信息
				logger.info("^^^^^^^^^^^^^^^^^^^ current view:"+ channel.getView());//查看当前组成员
				channel.getState(null, 1000);//状态同步，第一个参数为null则向协调者节点获取信息
			}
			
		}catch(Exception e){
			logger.error("^^^^^^^^^^^^^^^^^^^ genChannel exception: "+ e.getMessage());
			e.printStackTrace();
		}
	}
	

	/**  
	 * 当组成员发生变化的时候被调用  
	 */  
	@Override
	public void viewAccepted(View new_view) {
		logger.info("^^^^^^^^^^^^^^^^^^^ new view receiver : "+ new_view);
	}

	public void start() throws ChannelException{
		//genChannel();
	}
	
	public void sendMessage(String oper, String id, String key, Object value){
		try {
			if(channel == null)
				genChannel();
			
			StringBuffer msgBuffer = new StringBuffer(oper +",");
			if(id != null && id.trim().length() > 0)
				msgBuffer.append("id:"+ id);
			
			if(key != null && key.trim().length() > 0){
				msgBuffer.append(",");
				msgBuffer.append("key:"+ key);
			}
			
			if(value != null)
				msgBuffer.append(",value:"+ value);//应用层存储到会话中的对象的toString()方法不能用逗号分割属性值			
			
			if(msgBuffer != null && msgBuffer.toString().length() > 0){
				Message message = new Message(null, null, msgBuffer.toString());
				channel.send(message);//发送消息
				//channel.close();
			}
			
		} catch (ChannelNotConnectedException cnce) {
			cnce.printStackTrace();
		} catch (ChannelClosedException cce) {
			cce.printStackTrace();
		}catch (ChannelException ce) {
			ce.printStackTrace();
		}catch (Exception e) {
			e.printStackTrace();
		}
	}	
	
	private void operExpire(String sessionId){
		logger.info("^^^^^^^^^^^^^^^^^^^ Begin msg oper : "+ CacheConstants.OPER_KEY_EXPIRE);
		//logger.info("^^^^^^^^^^^^^^^^^^^ tomcat manager : "+ getManager());
		
		try{
			if(sessionId != null && sessionId.trim().length() > 0){
				Manager manager = getManager();
				Session session = manager.findSession(sessionId);
				if(session != null)
					manager.remove(session);
			}
			
		}catch(Exception e){
			logger.error("^^^^^^^^^^^^^^^^^^^ "+ CacheConstants.OPER_KEY_EXPIRE 
					+" exceptions : "+ e.getMessage());
		}
		
		logger.info("^^^^^^^^^^^^^^^^^^^ End msg oper : "+ CacheConstants.OPER_KEY_EXPIRE);
	}
	
	private void operAccess(String sessionId){
		logger.info("^^^^^^^^^^^^^^^^^^^ Begin msg oper : "+ CacheConstants.OPER_KEY_ACCESS);
		
		try{
			if(sessionId != null && sessionId.trim().length() > 0){
				Manager manager = getManager();
				Session session = manager.findSession(sessionId);
				if(session != null){
					CacheSession mySession = (CacheSession)session;
					logger.info("^^^^^^^^^^^^^^^^^^^ mySession : "+ mySession);
					
					mySession.lastAccessedTime = mySession.thisAccessedTime;
					mySession.thisAccessedTime = System.currentTimeMillis();
			        
			        if (mySession.ACTIVITY_CHECK) {
			        	mySession.accessCount.incrementAndGet();
			        }
				}
			}
			
		}catch(Exception e){
			logger.error("^^^^^^^^^^^^^^^^^^^ "+ CacheConstants.OPER_KEY_ACCESS 
					+" exceptions : "+ e.getMessage());
		}
		
		logger.info("^^^^^^^^^^^^^^^^^^^ End msg oper : "+ CacheConstants.OPER_KEY_ACCESS);
	}
	
	
	public static Manager getManager(){
		return manager;
	}
	public static void setManager(Manager _manager){
		manager = _manager;
	}
}