tomcat session复制(一)

lichaoyue888

　　tomcat的session复制大致分两种:all-to-all和backup，先看all-to-all,主要是记录下自己读源代码的心得和代码流程
　　tomcat集群配置暂时略过。
　　在tomcat的启动过程中，找到seesion复制的入口，在StandardContext的start()方法中:

            Manager contextManager = null;
if (manager == null) {
if ( (getCluster() != null) && distributable) {
try {
contextManager = getCluster().createManager(getName());
} catch (Exception ex) {
log.error("standardContext.clusterFail", ex);
ok = false;
}
} else {
contextManager = new StandardManager();
}
}
// Configure default manager if none was specified
if (contextManager != null) {
setManager(contextManager);
}
if (manager!=null && (getCluster() != null) && distributable) {
//let the cluster know that there is a context that is distributable
//and that it has its own manager
getCluster().registerManager(manager);
}
　　如果配置了tomcat的集群，那么 contextManager = getCluster().createManager(getName())就会被调用，返回的Manager的类型是DeltaManager，接着调用setManager(contextManager)方法：

    public synchronized void setManager(Manager manager) {
// Change components if necessary  管理session的manager
Manager oldManager = this.manager;
if (oldManager == manager)
return;
this.manager = manager;
// Stop the old component if necessary
if (started && (oldManager != null) &&
(oldManager instanceof Lifecycle)) {
try {
((Lifecycle) oldManager).stop();
} catch (LifecycleException e) {
log.error("ContainerBase.setManager: stop: ", e);
}
}
// Start the new component if necessary
if (manager != null)
manager.setContainer(this);
if (started && (manager != null) &&
(manager instanceof Lifecycle)) {
try {
((Lifecycle) manager).start();
} catch (LifecycleException e) {
log.error("ContainerBase.setManager: start: ", e);
}
}
// Report this property change to interested listeners
support.firePropertyChange("manager", oldManager, this.manager);
}

　　看到红色部分，这里调用了DeltaManager的start()方法，也就是启动了DeltaManager，看代码:

public void start() throws LifecycleException {
if (!initialized) init();
// Validate and update our current component state
if (started) {
return;
}
started = true;
lifecycle.fireLifecycleEvent(START_EVENT, null);
// Force initialization of the random number generator
generateSessionId();
// Load unloaded sessions, if any
try {
//the channel is already running
Cluster cluster = getCluster() ;
// stop remove cluster binding
//wow, how many nested levels of if statements can we have ;)
if(cluster == null) {
Container context = getContainer() ;
if(context != null && context instanceof Context) {
Container host = context.getParent() ;
if(host != null && host instanceof Host) {
cluster = host.getCluster();
if(cluster != null && cluster instanceof CatalinaCluster) {
setCluster((CatalinaCluster) cluster) ;
} else {
Container engine = host.getParent() ;
if(engine != null && engine instanceof Engine) {
cluster = engine.getCluster();
if(cluster != null && cluster instanceof CatalinaCluster) {
setCluster((CatalinaCluster) cluster) ;
}
} else {
cluster = null ;
}
}
}
}
}
if (cluster == null) {
log.error(sm.getString("deltaManager.noCluster", getName()));
return;
} else {
if (log.isInfoEnabled()) {
String type = "unknown" ;
if( cluster.getContainer() instanceof Host){
type = "Host" ;
} else if( cluster.getContainer() instanceof Engine){
type = "Engine" ;
}
log.info(sm.getString("deltaManager.registerCluster", getName(), type, cluster.getClusterName()));
}
}
if (log.isInfoEnabled()) log.info(sm.getString("deltaManager.startClustering", getName()));
//to survice context reloads, as only a stop/start is called, not
// createManager
cluster.registerManager(this);
//随机找一个节点，然后复制所有的session过来
getAllClusterSessions();



} catch (Throwable t) {
log.error(sm.getString("deltaManager.managerLoad"), t);
}
}
 　　首先，如果没有初始化，那么进行初始化，接着设置Cluster，然后调用Cluster的registerManager()方法，注册Manager，这里没有看明白，因为在StandardContext的start()方法中也有这个方法的调用，有重复的嫌疑？
　　最重要的是调用getAllClusterSessions()方法，这个方法的作用是，在集群中取到第一台tomcat的member，发送消息要求获得对方的所有session数据，对方收到消息，进行处理发送所有数据，这台tomcat收到数据，之后发送数据告知已经收到数据，双方通信结束。看代码：

    public synchronized void getAllClusterSessions() {
if (cluster != null && cluster.getMembers().length > 0) {
long beforeSendTime = System.currentTimeMillis();
   Member mbr = findSessionMasterMember();


if(mbr == null) { // No domain member found
return;
}
SessionMessage msg = new SessionMessageImpl(this.getName(),SessionMessage.EVT_GET_ALL_SESSIONS, null, "GET-ALL","GET-ALL-" + getName());
// set reference time
stateTransferCreateSendTime = beforeSendTime ;
// request session state
counterSend_EVT_GET_ALL_SESSIONS++;
stateTransfered = false ;
// FIXME This send call block the deploy thread, when sender waitForAck is enabled
try {
synchronized(receivedMessageQueue) {
receiverQueue = true ;
}
cluster.send(msg, mbr);


if (log.isWarnEnabled()) log.warn(sm.getString("deltaManager.waitForSessionState",getName(), mbr,getStateTransferTimeout()));
// FIXME At sender ack mode this method check only the state transfer and resend is a problem!
waitForSendAllSessions(beforeSendTime);


} finally {
synchronized(receivedMessageQueue) {
for (Iterator iter = receivedMessageQueue.iterator(); iter.hasNext();) {
SessionMessage smsg = (SessionMessage) iter.next();
if (!stateTimestampDrop) {
messageReceived(smsg, smsg.getAddress() != null ? (Member) smsg.getAddress() : null);
} else {
if (smsg.getEventType() != SessionMessage.EVT_GET_ALL_SESSIONS && smsg.getTimestamp() >= stateTransferCreateSendTime) {
// FIXME handle EVT_GET_ALL_SESSIONS later
messageReceived(smsg,smsg.getAddress() != null ? (Member) smsg.getAddress() : null);
} else {
if (log.isWarnEnabled()) {
log.warn(sm.getString("deltaManager.dropMessage",getName(), smsg.getEventTypeString(),new Date(stateTransferCreateSendTime), new Date(smsg.getTimestamp())));
}
}
}
}        
receivedMessageQueue.clear();
receiverQueue = false ;
}
}
} else {
if (log.isInfoEnabled()) log.info(sm.getString("deltaManager.noMembers", getName()));
}
}
　　首先，需要找到集群当中的一个tomcat，因为all-to-all复制中所有的tomcat都是对等的，所以第一个就可以了.
　　之后，生成一个EVT_GET_ALL_SESSIONS的消息，调用Manager的属性cluster的send()给集群中的tomcat发送要求得到所有session的消息，这个过程是异步的，所以在下面有waitForSendAllSessions()方法的调用，如果在60s内集群里面的tomcat没有给这个tomcat回应，那么这个tomcat就timeout，不能加入集群了。
　　我们先看SimpleTcpCluster的send()方法:

    public void send(ClusterMessage msg, Member dest) {
try {
msg.setAddress(getLocalMember());
if (dest != null) {
if (!getLocalMember().equals(dest)) {
channel.send(new Member[] {dest}, msg,channelSendOptions);


} else
log.error("Unable to send message to local member " + msg);
} else {
if (channel.getMembers().length>0)
channel.send(channel.getMembers(),msg,channelSendOptions);
else if (log.isDebugEnabled())
log.debug("No members in cluster, ignoring message:"+msg);
}
} catch (Exception x) {
log.error("Unable to send message through cluster sender.", x);
}
}
　　实际调用的是GroupChannel的send()方法，另外发送数据的类别是异步的(SEND_OPTIONS_ASYNCHRONOUS)
　　接着调用的是ChannelInterceptor的sendMessage()方法，这是个责任链模式的应用，其实默认的ChannelInterceptor只有两个：MessageDispatch15Interceptor和TcpFailureDetector。MessageDispatch15Interceptor只是在开了另外一个线程后，在一定条件下在该线程中进行sendMessage()方法调用。TcpFailureDetector的作用是处理sendMessage()方法出异常后对应的集群中的Member。最后面调用的ChannelCoordinator的sendMessage()方法：

    public void sendMessage(Member[] destination, ChannelMessage msg, InterceptorPayload payload) throws ChannelException {
if ( destination == null ) destination = membershipService.getMembers();
clusterSender.sendMessage(msg,destination);
if ( Logs.MESSAGES.isTraceEnabled() ) {
Logs.MESSAGES.trace("ChannelCoordinator - Sent msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " to "+Arrays.toNameString(destination));
}
}
 　　其实调用的是ParallelNioSender的sendMessage()方法，最终调用的是NioSender类进行底层的处理。到这里刚刚启动的tomcat的请求已经完成，这个时候在集群的中的一台tomcat收到消息，进行相应处理，接收消息的类是NioReceiver
　　让我们转换思想，转换到接收方进行考虑:
　　NioReceiver在前面的博文中已经详细讲过，其实他就是个后台线程，在不断的跑。在有消息进来，接收到后，每个消息产生一个Runnable任务，然后交给线程池处理。看Runnable任务的NioReplicationTask的drainChannel()方法中，有一段代码:

//process the message  ReceiverBase.messageDataReceived()
getCallback().messageDataReceived(msgs);
　　调用ListenCallback的方法，如下：

    public void messageDataReceived(ChannelMessage data) {
if ( this.listener != null ) {
if ( listener.accept(data) ) listener.messageReceived(data);
}
}
 　　这里的listener的实现类是:ChannelCoordinator,看代码是调用的父类的ChannelInterceptorBase的方法：

   public void messageReceived(ChannelMessage msg) {
if (getPrevious() != null) getPrevious().messageReceived(msg);
}
 　　这里的ChannelInterceptor还是前面说到的那两个，无关紧要，最终调用了GroupChannel的messageReceived()方法，具体的下一篇文章再说