怎样使用Memcached做Session同步

最近一直在看Xmemcached,希望能够通过它实现TOMCAT集群的Session复制,因为TOMCAT自己带的那个SESSION复制功能，当节点多了之后容易出错，且不稳定，还有就是不能同步Application里的东西，这几天捣鼓了很久，倒是把Application里的东西整合好了，但是让我头痛的是Session同步，不知道 这里有没有大侠做过类似的研究开发，希望能一起讨论下。
关于Application同步的代码我贴下：求指教

public class ClusterHelper {
	 
	private static final Logger logger = Logger.getLogger(ClusterHelper.class);
	private static String APPKEY = ApplicationHelper.APPKEY;
	private static boolean SysMemCached = ApplicationHelper.SysMemCached.equals("1")?true:false;
	
	public static void initMemcached(){  //系统首次启动的时候，初始化Memcached链接
		MemcachedClient memcachedClient = null;
		try{
			if(SysMemCached && ApplicationHelper.XMEMCACHEDCLIENT == null){
		         XMemcachedClientBuilder builder = new XMemcachedClientBuilder(AddrUtil.getAddresses("192.168.0.153:11211"));
		         builder.setConnectionPoolSize(50);
		         builder.addStateListener(new MemcachedListener());
		         builder.setCommandFactory(new BinaryCommandFactory());            
		         memcachedClient = builder.build();
		         memcachedClient.setConnectTimeout(20000);
		         memcachedClient.setMergeFactor(100);   //默认是150，缩小到50
		         memcachedClient.setOptimizeMergeBuffer(true);  //开启合并buffer的优化
		         memcachedClient.setEnableHeartBeat(false); //关闭心跳检测 暂时关闭  以后要开启
		         logger.debug("Create XMemcached Client successfully PoolSize:50 ConnectTimeout:20000ms !");
		         ApplicationHelper.XMEMCACHEDCLIENT = memcachedClient;
		         if(memcachedClient.get("CasHash")==null){
		        	 memcachedClient.add("CasHash", 0, new Hashtable());
		         }
		         if(ApplicationHelper.SysMenCasHash==null){
		        	 ApplicationHelper.SysMenCasHash = new Hashtable();
		         }
			}else{
				 //memcachedClient = ApplicationHelper.XMEMCACHEDCLIENT;
				 logger.debug("XMemcached Client Is Already In Application PoolSize:50 ConnectTimeout:20000ms !");
			}
	    }catch(Exception e){
	       e.printStackTrace();
	       ApplicationHelper.SysMemCached = "0";
	       logger.debug("Create XMemcached Client throws a Exception!");
	    }
	}
	
	
	/**
	 * 根据KEY检查对象在本地是否为最新(Application)
	 * @param mc
	 * @param key
	 * @return
	 */
	public static boolean isNewObj(MemcachedClient mc,String key){
		boolean flag = false;		
		try {
			String MApplicatonKey = APPKEY+key;
			Hashtable table = (Hashtable)mc.get("CasHash");  //记录在Memcached中存储的Application对象的版本号
			Hashtable Apptable = ApplicationHelper.SysMenCasHash;  //记录本节点上所有Application对象的版本号
			if(table==null || Apptable==null){
				return false;
			}			
			Long MemCas = (Long)table.get(MApplicatonKey);
			Long AppCas = (Long)Apptable.get(MApplicatonKey);
			if(MemCas==null || AppCas==null){
				return false;
			}
			
			logger.debug(((Long)table.get(MApplicatonKey)).longValue()==((Long)Apptable.get(MApplicatonKey)).longValue());
			
			if(((Long)table.get(MApplicatonKey)).longValue()==((Long)Apptable.get(MApplicatonKey)).longValue()){ //根据KEY比较本地与Memcached中的版本是否一致
				flag = true;
			}			
		} catch (TimeoutException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (MemcachedException e) {
			e.printStackTrace();
		}		
		return flag;
	}
	
	
	/**
	 * 更新本地及MEM上关于Application的CASHASH
	 * @param mc
	 * @param key
	 * @param value
	 */
	public static void setCasToHash(MemcachedClient mc,String key,long value){
		try {
			String MApplicatonKey = APPKEY+key;
			Hashtable table = (Hashtable)mc.get("CasHash");
			if(table.containsKey(MApplicatonKey)){
				table.remove(MApplicatonKey);				
			}
			table.put(MApplicatonKey, value);
			mc.replaceWithNoReply("CasHash", 0, table);
			
			Hashtable Apptable = ApplicationHelper.SysMenCasHash;
			if(Apptable.containsKey(MApplicatonKey)){
				Apptable.remove(MApplicatonKey);
			}
			Apptable.put(MApplicatonKey, value);
			ApplicationHelper.SysMenCasHash = Apptable;
			
		} catch (TimeoutException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (MemcachedException e) {
			e.printStackTrace();
		}
	}
	
	/**
	 * 根据key查看对象是否存在(Application)
	 * @param mc
	 * @param key
	 * @return
	 */
	public static boolean isExist(MemcachedClient mc,String key){
		boolean flag = false;
		try {
			if(((Hashtable)mc.get("CasHash")).containsKey(APPKEY+key)){
				flag = true;
			}
		} catch (TimeoutException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (MemcachedException e) {
			e.printStackTrace();
		}
		return flag;
	}
	
	/**
	 * 根据key删除对象(Application)
	 * @param mc
	 * @param key
	 */
	public static void ReMoveObj(MemcachedClient mc,String key){
		try {
			String MApplicatonKey = APPKEY+key; 
			Hashtable table = (Hashtable)mc.get("CasHash");
			if(table.containsKey(MApplicatonKey)){
				table.remove(MApplicatonKey);				
			}	
			mc.replaceWithNoReply("CasHash", 0, table);
			
			Hashtable Apptable = ApplicationHelper.SysMenCasHash;
			if(Apptable.containsKey(MApplicatonKey)){
				Apptable.remove(MApplicatonKey);
			}
			ApplicationHelper.SysMenCasHash = Apptable;
			
		} catch (TimeoutException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (MemcachedException e) {
			e.printStackTrace();
		}		
	}	
}

public class MemcachedServletContext{
	
	private static final Logger logger = Logger.getLogger(MemcachedServletContext.class);
    private MemcachedClient xmc = ApplicationHelper.XMEMCACHEDCLIENT;//全局XMC链接
    private boolean SysMemCached = ApplicationHelper.SysMemCached.equals("1")?true:false; //系统是否启用Memcached功能
    private String APPKEY = ApplicationHelper.APPKEY; 
	private ServletContext application;
	
	public MemcachedServletContext(ServletContext application){
		this.application = application;
	}
	
	public Object getAttribute(String arg0) {
		Object obj = null;
		if(SysMemCached && !ClusterHelper.isNewObj(xmc, arg0)){	
	        try {
	        	GetsResponse GRObj = xmc.gets(APPKEY+arg0);
	        	
	        	if(GRObj == null) return null;
	        	
	        	obj = GRObj.getValue();
	        	
	        	application.setAttribute(arg0,obj);
	        	
				Hashtable Apptable = ApplicationHelper.SysMenCasHash;
				if(Apptable.containsKey(APPKEY+arg0)){
					Apptable.remove(APPKEY+arg0);
				}
				
				Apptable.put(APPKEY+arg0,GRObj.getCas());
			} catch (TimeoutException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (MemcachedException e) {
				e.printStackTrace();
			}			
		}else{
			logger.debug("++++++++++++++++++++++++++");
			return application.getAttribute(arg0);			
		}
        return obj;
	}
	
	public void removeAttribute(String arg0) {
		if(SysMemCached){		
	        try {
				xmc.deleteWithNoReply(APPKEY+arg0);
				ClusterHelper.ReMoveObj(xmc,arg0);
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (MemcachedException e) {
				e.printStackTrace();
			}			
		}		
		application.removeAttribute(arg0);			
	}

	
	public void setAttribute(String arg0, Object arg1) {
		removeAttribute(arg0);
		application.setAttribute(arg0, arg1);
		if(SysMemCached && arg1 instanceof java.io.Serializable){
			try {
				if(ClusterHelper.isExist(xmc, arg0)){
				    //xmc.replaceWithNoReply(APPKEY+arg0, 0, arg1);
					new CASThread(xmc,0,APPKEY+arg0,arg1,new CountDownLatch(1)).start();
				    logger.debug("replace::"+APPKEY+arg0);
				}else{
				    xmc.addWithNoReply(APPKEY+arg0, 0, arg1);
				    logger.debug("add::"+APPKEY+arg0);
				}			
				ClusterHelper.setCasToHash(xmc,arg0,(xmc.gets(APPKEY+arg0)).getCas());
			} catch (TimeoutException e) {
				e.printStackTrace();
			} catch (InterruptedException e) {
				e.printStackTrace();
			} catch (MemcachedException e) {
				e.printStackTrace();
			}
		}
	}

}

评论

1 楼 wqy_blue3316 2011-07-08  

自己顶下，难道没有人会么～