tomcat session复制(一)
?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[i]);
?调用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()方法，具体的下一篇文章再说
?
?
?
?