tomcat session复制(二)

      上文http://nod0620.iteye.com/admin/blogs/1030398 写了session复制的发送部分，继续接收部分：

当接收方tomcat接收到需要session的消息时，最终调用了GroupChannel的messageReceived()方法

    public void messageReceived(ChannelMessage msg) {
        if ( msg == null ) return;
        try {
            if ( Logs.MESSAGES.isTraceEnabled() ) {
                Logs.MESSAGES.trace("GroupChannel - Received msg:" + new UniqueId(msg.getUniqueId()) + " at " +new java.sql.Timestamp(System.currentTimeMillis())+ " from "+msg.getAddress().getName());
            }

            Serializable fwd = null;
            if ( (msg.getOptions() & SEND_OPTIONS_BYTE_MESSAGE) == SEND_OPTIONS_BYTE_MESSAGE ) {
                fwd = new ByteMessage(msg.getMessage().getBytes());
            } else {
                try {
                    fwd = XByteBuffer.deserialize(msg.getMessage().getBytesDirect(), 0, msg.getMessage().getLength());
                }catch (Exception sx) {
                    log.error("Unable to deserialize message:"+msg,sx);
                    return;
                }
            }
            if ( Logs.MESSAGES.isTraceEnabled() ) {
                Logs.MESSAGES.trace("GroupChannel - Receive Message:" + new UniqueId(msg.getUniqueId()) + " is " +fwd);
            }

            //get the actual member with the correct alive time
            Member source = msg.getAddress();
            boolean rx = false;
            boolean delivered = false;
            for ( int i=0; i<channelListeners.size(); i++ ) {
                ChannelListener channelListener = (ChannelListener)channelListeners.get(i);
                if (channelListener != null && channelListener.accept(fwd, source)) {
                    channelListener.messageReceived(fwd, source);
                    delivered = true;
                    //if the message was accepted by an RPC channel, that channel
                    //is responsible for returning the reply, otherwise we send an absence reply
                    if ( channelListener instanceof RpcChannel ) rx = true;
                }
            }//for
            if ((!rx) && (fwd instanceof RpcMessage)) {
                //if we have a message that requires a response,
                //but none was given, send back an immediate one
                sendNoRpcChannelReply((RpcMessage)fwd,source);
            }
            if ( Logs.MESSAGES.isTraceEnabled() ) {
                Logs.MESSAGES.trace("GroupChannel delivered["+delivered+"] id:"+new UniqueId(msg.getUniqueId()));
            }

        } catch ( Exception x ) {
            //this could be the channel listener throwing an exception, we should log it 
            //as a warning.
            if ( log.isWarnEnabled() ) log.warn("Error receiving message:",x);
            throw new RemoteProcessException("Exception:"+x.getMessage(),x);
        }
    }

    可以看到，这里主要是调用ChannelListener的messageReceived方法，显然这个实现类就是SimpleTcpCluster：

 

 public void messageReceived(Serializable message, Member sender) {
        ClusterMessage fwd = (ClusterMessage)message;
        fwd.setAddress(sender);
        messageReceived(fwd);
    }

    public void messageReceived(ClusterMessage message) {

        long start = 0;
        if (log.isDebugEnabled() && message != null)
            log.debug("Assuming clocks are synched: Replication for "
                    + message.getUniqueId() + " took="
                    + (System.currentTimeMillis() - (message).getTimestamp())
                    + " ms.");

        //invoke all the listeners
        boolean accepted = false;
        if (message != null) {
            for (Iterator iter = clusterListeners.iterator(); iter.hasNext();) {
                ClusterListener listener = (ClusterListener) iter.next();
                if (listener.accept(message)) {
                    accepted = true;
                    listener.messageReceived(message);
                }
            }
        }
        if (!accepted && log.isDebugEnabled()) {
            if (notifyLifecycleListenerOnFailure) {
                Member dest = message.getAddress();
                // Notify our interested LifecycleListeners
                lifecycle.fireLifecycleEvent(RECEIVE_MESSAGE_FAILURE_EVENT,
                        new SendMessageData(message, dest, null));
            }
            log.debug("Message " + message.toString() + " from type "
                    + message.getClass().getName()
                    + " transfered but no listener registered");
        }
        return;
    }

   其实这里调用的是ClusterListener的messageReceived()方法，默认的两个ClusterListener是ClusterSessionListener和JvmRouteSessionIDBinderListener，JvmRouteSessionIDBinderListener只有在发送的消息是关于session  id改变的消息时才起作用，重点看ClusterSessionListener.messageReceived()方法，其实是调用到了DeltaManager的messageDataReceived()方法：

    public void messageDataReceived(ClusterMessage cmsg) {
        if (cmsg != null && cmsg instanceof SessionMessage) {
            SessionMessage msg = (SessionMessage) cmsg;
            switch (msg.getEventType()) {
                case SessionMessage.EVT_GET_ALL_SESSIONS:
                case SessionMessage.EVT_SESSION_CREATED: 
                case SessionMessage.EVT_SESSION_EXPIRED: 
                case SessionMessage.EVT_SESSION_ACCESSED:
                case SessionMessage.EVT_SESSION_DELTA:
                case SessionMessage.EVT_CHANGE_SESSION_ID: {
                    synchronized(receivedMessageQueue) {
                        if(receiverQueue) {
                            receivedMessageQueue.add(msg);
                            return ;
                        }
                    }
                   break;
                }
                default: {
                    //we didn't queue, do nothing
                    break;
                }
            } //switch
            
            messageReceived(msg, msg.getAddress() != null ? (Member) msg.getAddress() : null);
        }
    }

 

     这里的事件是EVT_GET_ALL_SESSIONS，在messageReceived()方法中有这个分支的处理代码，最后面调用

handleGET_ALL_SESSIONS()方法：

 protected void handleGET_ALL_SESSIONS(SessionMessage msg, Member sender) throws IOException {
        counterReceive_EVT_GET_ALL_SESSIONS++;
        //get a list of all the session from this manager
        if (log.isDebugEnabled()) log.debug(sm.getString("deltaManager.receiveMessage.unloadingBegin", getName()));
        // Write the number of active sessions, followed by the details
        // get all sessions and serialize without sync
        Session[] currentSessions = findSessions();
        long findSessionTimestamp = System.currentTimeMillis() ;
        if (isSendAllSessions()) {
            sendSessions(sender, currentSessions, findSessionTimestamp);
        } else {
            // send session at blocks
            for (int i = 0; i < currentSessions.length; i += getSendAllSessionsSize()) {
                int len = i + getSendAllSessionsSize() > currentSessions.length ? currentSessions.length - i : getSendAllSessionsSize();
                Session[] sendSessions = new Session[len];
                System.arraycopy(currentSessions, i, sendSessions, 0, len);
                sendSessions(sender, sendSessions,findSessionTimestamp);
                if (getSendAllSessionsWaitTime() > 0) {
                    try {
                        Thread.sleep(getSendAllSessionsWaitTime());
                    } catch (Exception sleep) {
                    }
                }//end if
            }//for
        }//end if
        
        SessionMessage newmsg = new SessionMessageImpl(name,SessionMessage.EVT_ALL_SESSION_TRANSFERCOMPLETE, null,"SESSION-STATE-TRANSFERED", "SESSION-STATE-TRANSFERED"+ getName());
        newmsg.setTimestamp(findSessionTimestamp);
        if (log.isDebugEnabled()) log.debug(sm.getString("deltaManager.createMessage.allSessionTransfered",getName()));
        counterSend_EVT_ALL_SESSION_TRANSFERCOMPLETE++;
        cluster.send(newmsg, sender);
    }

   主要是先找到所有的session，然后发送所有的session，然后发送一个标志结束的消息表示发送完成。这样第一次tomcat的启动加入集群是拿所用的session就完成了，主要的流程是这样的：

 

发送方：

StandardContext.start()--->SimpleTcpCluster.createManager()--->StandardContext.setManager()--->

DeltaManager.start()--->DeltaManager.getAllClusterSessions()--->SimpleTcpCluster.send()

---->GroupChannel.send()---->ParallelNioSender.sendMessage()--->NioSender.process()

 

接收方：

NioReceiver.listen()--->NioReceiver.readDataFromSocket()---->NioReplicationTask.drainChannel()--->

ListenCallback.messageDataReceived()---->ChannelCoordinator.messageReceived()--->

GroupChannel.messageReceived()--->SimpleTcpCluster.messageReceived()--->

DeltaManager.messageDataReceived()---->DeltaManager.handleGET_ALL_SESSIONS()

 

接收方准备好数据又要发送给发送方，简单来说通信是这样：

发送方发送请求--->接收方接受请求，发送数据，一个是session的数据，一个是session发送完毕的数据--->

发送方收到接收方过来的两种数据