【原创】Tomcat集群环境下对session进行外部缓存的方法（2）

wdx1992828

　　Session对象的持久化比较麻烦，虽然有序列化，但是并不确定Session对象中保存的其他信息是否可以序列化，这可能是网上很多解决方案摒弃此种做法的原因，网上的很多做法都是将Session中的attribute信息持久化并结构化存储，这显然很方便，但是session中的其他信息就丢了，否则仍然占据中间件内存，通过查看源码，惊喜的发现Tomcat对象提供了Session序列化的接口以及相关实现（Store），不过不是很满足需求，对其进行了一些改造就ok了，最终，Session对象作为一个整体，以二进制的形式保存在blob中，当反序列化时，还要装配Session和SessionManager之间的依赖关系。主要用到了session的以下方法，主要思想就是将session化整为零，传入一个输入流，将属性都写入该流中：



1  protected void writeObject(ObjectOutputStream stream) throws IOException {
2
3         // Write the scalar instance variables (except Manager)
4         stream.writeObject(new Long(creationTime));
5         stream.writeObject(new Long(lastAccessedTime));
6         stream.writeObject(new Integer(maxInactiveInterval));
7         stream.writeObject(new Boolean(isNew));
8         stream.writeObject(new Boolean(isValid));
9         stream.writeObject(new Long(thisAccessedTime));
10         stream.writeObject(id);
11         if (manager.getContainer().getLogger().isDebugEnabled())
12             manager.getContainer().getLogger().debug
13                 ("writeObject() storing session " + id);
14
15         // Accumulate the names of serializable and non-serializable attributes
16         String keys[] = keys();
17         ArrayList saveNames = new ArrayList();
18         ArrayList saveValues = new ArrayList();
19         for (int i = 0; i < keys.length; i++) {
20             Object value = attributes.get(keys);
21             if (value == null)
22                 continue;
23             else if ( (value instanceof Serializable)
24                     && (!exclude(keys) )) {
25                 saveNames.add(keys);
26                 saveValues.add(value);
27             } else {
28                 removeAttributeInternal(keys, true);
29             }
30         }
31
32         // Serialize the attribute count and the Serializable attributes
33         int n = saveNames.size();
34         stream.writeObject(new Integer(n));
35         for (int i = 0; i < n; i++) {
36             stream.writeObject((String) saveNames.get(i));
37             try {
38                 stream.writeObject(saveValues.get(i));
39                 if (manager.getContainer().getLogger().isDebugEnabled())
40                     manager.getContainer().getLogger().debug
41                         ("  storing attribute '" + saveNames.get(i) +
42                         "' with value '" + saveValues.get(i) + "'");
43             } catch (NotSerializableException e) {
44                 manager.getContainer().getLogger().warn
45                     (sm.getString("standardSession.notSerializable",
46                      saveNames.get(i), id), e);
47                 stream.writeObject(NOT_SERIALIZED);
48                 if (manager.getContainer().getLogger().isDebugEnabled())
49                     manager.getContainer().getLogger().debug
50                        ("  storing attribute '" + saveNames.get(i) +
51                         "' with value NOT_SERIALIZED");
52             }
53         }
54
55     }
　　这样将这个流的数据转换为一个字节流保存为blob即可。还原时，同样使用以下方法还原，将该流传入，方法会返回一个离线的Session对象，什么叫离散的？就是没有和具体的SessionMananger关联的：



1 protected void readObject(ObjectInputStream stream)
2         throws ClassNotFoundException, IOException {
3
4         // Deserialize the scalar instance variables (except Manager)
5         authType = null;        // Transient only
6         creationTime = ((Long) stream.readObject()).longValue();
7         lastAccessedTime = ((Long) stream.readObject()).longValue();
8         maxInactiveInterval = ((Integer) stream.readObject()).intValue();
9         isNew = ((Boolean) stream.readObject()).booleanValue();
10         isValid = ((Boolean) stream.readObject()).booleanValue();
11         thisAccessedTime = ((Long) stream.readObject()).longValue();
12         principal = null;        // Transient only
13         //        setId((String) stream.readObject());
14         id = (String) stream.readObject();
15         if (manager.getContainer().getLogger().isDebugEnabled())
16             manager.getContainer().getLogger().debug
17                 ("readObject() loading session " + id);
18
19         // Deserialize the attribute count and attribute values
20         if (attributes == null)
21             attributes = new Hashtable();
22         int n = ((Integer) stream.readObject()).intValue();
23         boolean isValidSave = isValid;
24         isValid = true;
25         for (int i = 0; i < n; i++) {
26             String name = (String) stream.readObject();
27             Object value = (Object) stream.readObject();
28             if ((value instanceof String) && (value.equals(NOT_SERIALIZED)))
29                 continue;
30             if (manager.getContainer().getLogger().isDebugEnabled())
31                 manager.getContainer().getLogger().debug("  loading attribute '" + name +
32                     "' with value '" + value + "'");
33             attributes.put(name, value);
34         }
35         isValid = isValidSave;
36
37         if (listeners == null) {
38             listeners = new ArrayList();
39         }
40
41         if (notes == null) {
42             notes = new Hashtable();
43         }
44     }
　　
　　说说我为什么实现以上三种缓存方案。
　　最开始我通过Map缓存（SessionCacheMap）测试通过之后，立刻将其迁移到MemCache中，因为Map缓存还是属于JVM进程内缓存，Session仍然在中间件内部，只是保存在我自定义的一块内存区域中而已，只是验证是否可以完全的拦截，因此没有实际意义，需要将其从JVM内部拿出来。因此我迁移至MemCache中去了。
　　而为什么最后没有使用MemCache，MemCache没有保障，这个缓存性能很高，功能强大，但是它并不对缓存信息持久化，一旦它宕掉，session信息就都丢了，如果一些页面缓存丢了倒也没什么，重启再建立就好了，但是如果session信息丢了，那就比较严重了，而且这个缓存没有持久化方案，所以，最后我没有采用它缓存session，而是使用它来缓存页面。
　　所以我选择了dbms来存储session信息，dbms强大、可靠的数据管理保证session不会丢失，每个session都被持久化进数据库中，上个例子我们可以看到，我们把所有节点都关闭再启动，客户端刷新界面，仍然可以正常显示，session不会丢失。
　　可靠性有了保证，获取session是一个非常频繁的操作，如何保证性能呢？每次都去查询数据库，性能还是没有保证，这地方很容易成为瓶颈，尤其大规模访问时，虽然oracle可以将记录保存在buffercache中，但是毕竟它不是专业的缓存， bufferCache大小有限，而且每次插入删除session信息会涉及到磁盘的读写，这都是瓶颈，所以，我又引入了timesten内存数据库，它在Oracle的前端作为Oracle的一个超大的bufferCache，session持久化信息首先进入timesten，而timesten后台异步将session持久化至oracle中， TimeSten其实就是一个大的前端缓存而已，进一步讲，就是一个外置的bufferCache，后端的Oracle保证TimeSten万一宕掉，持久化的session数据不会丢失，宕掉就宕掉了，重启就ok了。所以，我在TimeSten11g版本下搭建了环境，建立了一个AWT类型的CacheGroup这个Cache就缓存了一张表名为T_SESSION：这张表的定义为：



1 create table T_SESSION
2 (
3   C_SID     VARCHAR2(200) primary key ,
4   C_SESSION BLOB
5 )
　　基于这张表的CacheGroup我定义如下（oracle的blob在timesten中影射为varbinary）：



create dynamic asynchronous writethrough cache group g_awt from uss.t_session ( c_sid varchar(200) not null , c_session varbinary(262144),primary key(c_sid));
　　该Group采用异步方式写入oracle，数据的插入、删除都在timesten中完成，而timesten会异步的将数据刷新至oracle，看下例子：
　　首先在tt中插入1条数据：

　　再看下oracle中：

　　Timesten中删掉一条记录：

　　再看oracle中：

　　发现数据已经被异步更新。因此，采用timesten缓存session，oracle持久化session的基础环境已经OK了。下面其实就是将原本往oracle中保存的session往timesten中保存即可，其实就是切换一下Connection来源即可，此处省略。

　　那么如何处理session失效的问题？tomcat的session有一个lifescycle的概念，但是现在我把session对象从tomcat中完全剥离出来了，这样便不能由tomcat来维护了，怎么办呢？其实很简单，方案有两种：
　　1.我在t_session表的后面增加了d_create一列，类型为timestamp，该列代表这个session对象的创建时间戳，当session创建时记录时间戳，当session被修改时，更新这个时间戳，我定义一个job，这个job每5秒钟扫描一遍t_session记录，将截至到现在d_create超过某值的记录删除：



Delete from t_session where sysdate-c_create>n;这个n就是session的失效时间。
　　2.如果使用内存数据库，那更简单了，在文档中发现timesten有aging的概念，什么叫aging？看下面的CacheGroup定义：



create dynamic asynchronous writethrough cache group g_awt from uss.t_session ( c_sid varchar(200) not null , c_session varbinary(262144),d_create timestamp,primary key(c_sid)) AGING USE  d_create LIFETIME 15 minutes CYCLE 5 seconds ON;
　　aging子句代表每隔6秒钟检测一次记录，使d_create超过15分钟的记录失效，是不是很方便？交给timesten去维护就ok了。
　　

　　再说说我是如何实现页面缓存的，页面缓存使用Filter来实现，包装了ResponseWrapper类，该类拦截response的信息，将中间件的响应数据拦截并保存至MemCache中，当下次同样的URL到来，只需要去MemCache中取就可以了。

　　PageCacheFilter用于拦截url，将指定规则的url拦截响应内容并缓存至MemCache，其中具体的是由MonitorResponseWrapper类来完成的，中间件的响应数据存储在它的ByteArray输出流中，它继承自HttpServletResponseWrapper类,典型的装饰模式的应用。



package com.thunisoft.filter;
import java.io.IOException;
import java.util.Date;
import javax.servlet.Filter;
import javax.servlet.FilterChain;
import javax.servlet.FilterConfig;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
public class PageCacheFilter implements Filter {
public void destroy() {
// TODO Auto-generated method stub

}
public void doFilter(ServletRequest req, ServletResponse resp,
FilterChain chain) throws IOException, ServletException {
MonitorResponseWrapper response = new MonitorResponseWrapper(
(HttpServletResponse) resp);
String url = ((HttpServletRequest) req).getRequestURI();
// MemCachedManager.getInstance().delete(url);
Object o = MemCachedManager.getInstance().get(url);
Date d = new Date(System.currentTimeMillis() + 10000);
if (o == null) {
chain.doFilter(req, response);
byte b[] = response.getResponseData();
resp.getOutputStream().write(b);
MemCachedManager.getInstance().set(url, b, d);
} else {
byte[] b = (byte[]) o;
resp.getOutputStream().write(b);
}
}
public static void main(String[] args) {
Date d = new Date(500);
}
public void init(FilterConfig arg0) throws ServletException {
// TODO Auto-generated method stub

}
}


package com.thunisoft.filter;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.OutputStreamWriter;
import java.io.PrintWriter;
import java.io.UnsupportedEncodingException;
import javax.servlet.ServletOutputStream;
import javax.servlet.http.HttpServletResponse;
import javax.servlet.http.HttpServletResponseWrapper;
public class MonitorResponseWrapper extends HttpServletResponseWrapper {
private ByteArrayOutputStream buffer = null;
private ServletOutputStream out = null;
private PrintWriter writer = null;
public MonitorResponseWrapper(HttpServletResponse resp) throws IOException {
super(resp);
buffer = new ByteArrayOutputStream();// 真正存储数据的流
out = new WapperedOutputStream(buffer);
writer = new PrintWriter(new OutputStreamWriter(buffer, this
.getCharacterEncoding()));
}
// 重载父类获取outputstream的方法
    @Override
public ServletOutputStream getOutputStream() throws IOException {
return out;
}
// 重载父类获取writer的方法
    @Override
public PrintWriter getWriter() throws UnsupportedEncodingException {
return writer;
}
// 重载父类获取flushBuffer的方法
    @Override
public void flushBuffer() throws IOException {
if (out != null) {
out.flush();
}
if (writer != null) {
writer.flush();
}
}
static int i;
public static void main(String[] args) {
System.out.println(i);
}
@Override
public void reset() {
buffer.reset();
}
public byte[] getResponseData() throws IOException {
// 将out、writer中的数据强制输出到WapperedResponse的buffer里面，否则取不到数据
        flushBuffer();
return buffer.toByteArray();
}
// 内部类，对ServletOutputStream进行包装
private class WapperedOutputStream extends ServletOutputStream {
private ByteArrayOutputStream bos = null;
public WapperedOutputStream(ByteArrayOutputStream stream)
throws IOException {
bos = stream;
}
@Override
public void write(int b) throws IOException {
bos.write(b);
}
}
}


package com.thunisoft.filter;
import java.util.Date;
import org.apache.catalina.Session;
import com.danga.MemCached.MemCachedClient;
import com.danga.MemCached.SockIOPool;
import com.thunisoft.session.CachedSession;
/**
*
* Memcached 缓存管理
*
* @author Administrator
*
*
*/
public class MemCachedManager {
// 创建全局的唯一实例
protected static MemCachedClient mcc = new MemCachedClient();
protected static MemCachedManager memCachedManager = new MemCachedManager();
// 设置与缓存服务器的连接池
static {
// 服务器列表和其权重
String[] servers = { "172.20.70.251:12000" };
Integer[] weights = { 3 };
// 获取socke连接池的实例对象
SockIOPool pool = SockIOPool.getInstance();
// 设置服务器信息
        pool.setServers(servers);
pool.setWeights(weights);
// 设置初始连接数、最小和最大连接数以及最大处理时间
pool.setInitConn(5);
pool.setMinConn(5);
pool.setMaxConn(250);
pool.setMaxIdle(1000 * 60 * 60 * 6);
// 设置主线程的睡眠时间
pool.setMaintSleep(30);
// 设置TCP的参数，连接超时等
pool.setNagle(false);
pool.setSocketTO(3000);
pool.setSocketConnectTO(0);
// 初始化连接池
        pool.initialize();
// 压缩设置，超过指定大小（单位为K）的数据都会被压缩
mcc.setCompressEnable(true);
mcc.setCompressThreshold(64 * 1024);
}
/**
*
* 保护型构造方法，不允许实例化！
*
*
*/
protected MemCachedManager() {
}
/**
*
* 获取唯一实例.
*
*
*
* @return
*/
public synchronized static MemCachedManager getInstance() {
return memCachedManager;
}
/**
*
* 添加一个指定的值到缓存中.
*
*
*
* @param key
*
* @param value
*
* @return
*/
public boolean add(String key, Object value) {
return mcc.add(key, value);
}
public boolean add(String key, Object value, Date expiry) {
return mcc.add(key, value, expiry);
}
public boolean set(String key, Object value, Date expiry) {
return mcc.set(key, value, expiry);
}
public boolean set(String key, Object value) {
return mcc.set(key, value);
}
/**
*
* 更新缓存对象
*
* @param key
*
* @param value
*
* @return
*/
public boolean replace(String key, Object value) {
return mcc.replace(key, value);
}
public boolean replace(String key, Object value, Date expiry) {
return mcc.replace(key, value, expiry);
}
/**
*
* 根据指定的关键字获取对象.
*
*
*
* @param key
*
* @return
*/
public Object get(String key) {
return mcc.get(key);
}
public boolean delete(String key){
return mcc.delete(key);
}
public static void main(String[] args) {
MemCachedManager cache = MemCachedManager.getInstance();
Session s=new CachedSession(null,null);
cache.set("111",s, new Date(System.currentTimeMillis()+10*1000));
//        Object o=cache.get("E4A8127405876F0F94627BB0F440CCDC");
        
System.out.println("get value : " + cache.get("111"));
}
}



package com.thunisoft.filter;
import java.util.Date;
import org.apache.catalina.Session;
import com.danga.MemCached.MemCachedClient;
import com.danga.MemCached.SockIOPool;
import com.thunisoft.session.CachedSession;
/**
*
* Memcached 缓存管理
*
* @author Administrator
*
*
*/
public class MemCachedManager {
// 创建全局的唯一实例
protected static MemCachedClient mcc = new MemCachedClient();
protected static MemCachedManager memCachedManager = new MemCachedManager();
// 设置与缓存服务器的连接池
static {
// 服务器列表和其权重
String[] servers = { "172.20.70.251:12000" };
Integer[] weights = { 3 };
// 获取socke连接池的实例对象
SockIOPool pool = SockIOPool.getInstance();
// 设置服务器信息
        pool.setServers(servers);
pool.setWeights(weights);
// 设置初始连接数、最小和最大连接数以及最大处理时间
pool.setInitConn(5);
pool.setMinConn(5);
pool.setMaxConn(250);
pool.setMaxIdle(1000 * 60 * 60 * 6);
// 设置主线程的睡眠时间
pool.setMaintSleep(30);
// 设置TCP的参数，连接超时等
pool.setNagle(false);
pool.setSocketTO(3000);
pool.setSocketConnectTO(0);
// 初始化连接池
        pool.initialize();
// 压缩设置，超过指定大小（单位为K）的数据都会被压缩
mcc.setCompressEnable(true);
mcc.setCompressThreshold(64 * 1024);
}
/**
*
* 保护型构造方法，不允许实例化！
*
*
*/
protected MemCachedManager() {
}
/**
*
* 获取唯一实例.
*
*
*
* @return
*/
public synchronized static MemCachedManager getInstance() {
return memCachedManager;
}
/**
*
* 添加一个指定的值到缓存中.
*
*
*
* @param key
*
* @param value
*
* @return
*/
public boolean add(String key, Object value) {
return mcc.add(key, value);
}
public boolean add(String key, Object value, Date expiry) {
return mcc.add(key, value, expiry);
}
public boolean set(String key, Object value, Date expiry) {
return mcc.set(key, value, expiry);
}
public boolean set(String key, Object value) {
return mcc.set(key, value);
}
/**
*
* 更新缓存对象
*
* @param key
*
* @param value
*
* @return
*/
public boolean replace(String key, Object value) {
return mcc.replace(key, value);
}
public boolean replace(String key, Object value, Date expiry) {
return mcc.replace(key, value, expiry);
}
/**
*
* 根据指定的关键字获取对象.
*
*
*
* @param key
*
* @return
*/
public Object get(String key) {
return mcc.get(key);
}
public boolean delete(String key){
return mcc.delete(key);
}
public static void main(String[] args) {
MemCachedManager cache = MemCachedManager.getInstance();
Session s=new CachedSession(null,null);
cache.set("111",s, new Date(System.currentTimeMillis()+10*1000));
//        Object o=cache.get("E4A8127405876F0F94627BB0F440CCDC");
        
System.out.println("get value : " + cache.get("111"));
}
}
　　所以，最终，这个方案的整体架构如下：

　　这个架构的优点有以下：
　　使用廉价的Tomcat集群，摒弃了效率低下的session复制，可以实现大规模的横向扩展，增加网站的吞吐量和并发访问量，如果网站的并发访问超大，那么可以对前端的Apache进行分层的扩展，也很简单，未来Session如果更多的话，那Timsten的查询可能会成为瓶颈，那可以继续将TimeSten拆分，在tomcat的会话管理器中对Session查询进行路由，比如对sessionid进行hash散列，将session分散缓存至分布式的TimeSten中，客户端请求时，根据hash映射确定sessionid存在于哪个TimeSten节点中，达到分散压力的目的，这只是对于一般网站，如果对于mis系统，可能最终的压力都跑到数据库上而非中间件了，那又是另外一回事了。