tomcat NioSender

      上次说了NioReceiver，这次看看NioSender，NioSender的体系结构比较复杂的，但感觉不用这么复杂，先看下类图：

 

       这里觉得ParallelNioSender和PooledParallelSender之间有一个类可以去掉，如去了PooledParallelSender，PooledSender类设计成接口，这样的话会比较简洁.

 

ChannelCoordinator使用的是PooledParallelSender，最后面最重要的是还是NioSender。

 

NioSender是个状态机，看注释：

 * - NOT_CONNECTED -> connect() -> CONNECTED
 * - CONNECTED -> setMessage() -> READY TO WRITE
 * - READY_TO_WRITE -> write() -> READY TO WRITE | READY TO READ
 * - READY_TO_READ -> read() -> READY_TO_READ | TRANSFER_COMPLETE
 * - TRANSFER_COMPLETE -> CONNECTED

 这样可以循环的利用：

 public boolean process(SelectionKey key, boolean waitForAck) throws IOException {
        int ops = key.readyOps();
        key.interestOps(key.interestOps() & ~ops);
        //in case disconnect has been called
        if ((!isConnected()) && (!connecting)) throw new IOException("Sender has been disconnected, can't selection key.");
        if ( !key.isValid() ) throw new IOException("Key is not valid, it must have been cancelled.");
        if ( key.isConnectable() ) {
            if ( socketChannel.finishConnect() ) {
                completeConnect();
                if ( current != null ) key.interestOps(key.interestOps() | SelectionKey.OP_WRITE);
                return false;
            } else  { 
                //wait for the connection to finish
                key.interestOps(key.interestOps() | SelectionKey.OP_CONNECT);
                return false;
            }//end if
        } else if ( key.isWritable() ) {
            boolean writecomplete = write(key);
            if ( writecomplete ) {
                //we are completed, should we read an ack?
                if ( waitForAck ) {
                    //register to read the ack
                    key.interestOps(key.interestOps() | SelectionKey.OP_READ);
                } else {
                    //if not, we are ready, setMessage will reregister us for another write interest
                    //do a health check, we have no way of verify a disconnected
                    //socket since we don't register for OP_READ on waitForAck=false
                    read(key);//this causes overhead
                    setRequestCount(getRequestCount()+1);
                    return true;
                }
            } else {
                //we are not complete, lets write some more
                key.interestOps(key.interestOps()|SelectionKey.OP_WRITE);
            }//end if
        } else if ( key.isReadable() ) {
            boolean readcomplete = read(key);
            if ( readcomplete ) {
                setRequestCount(getRequestCount()+1);
                return true;
            } else {
                key.interestOps(key.interestOps() | SelectionKey.OP_READ);
            }//end if
        } else {
            //unknown state, should never happen
            log.warn("Data is in unknown state. readyOps="+ops);
            throw new IOException("Data is in unknown state. readyOps="+ops);
        }//end if
        return false;
    }

   首先屏蔽了SelectionKey的interest集合中的ready集合，只有在这一次处理完以后才会恢复，这样的话下一次的数据不会影响到这一次的数，之后就是 SelectionKey的各种状态的处理，其中read()是调用write()后用于接受ack回执的.

 

 

  从ChannelCoordinator开始，来看看NioSender的调用的流程：

 

ChannelCoordinator.sendMessage()---->ReplicationTransmitter.sendMessage()---->PooledParallelSender.sendMessage()

---->ParallelNioSender.sendMessage()---->ParallelNioSender.doLoop()---->NioSender.process()

 

可以吧PooledParallelSender看成一个门面，在这里是NioSender的入口，PooledSender里面主要是一个资源池的作用，这样可以合理利用和控制资源。

看PooledParallelSender的sendMessage()方法:

if ( !connected ) throw new ChannelException("Sender not connected.");
        ParallelNioSender sender = (ParallelNioSender)getSender();
        if (sender == null) {
            ChannelException cx = new ChannelException("Unable to retrieve a data sender, time out error.");
            for (int i = 0; i < destination.length; i++) cx.addFaultyMember(destination[i], new NullPointerException("Unable to retrieve a sender from the sender pool"));
            throw cx;
        } else {
            try {
                sender.sendMessage(destination, message);
                sender.keepalive();
            } catch (ChannelException x) {
                sender.disconnect();
                throw x;
            } finally {
                returnSender(sender);
                if (!connected) disconnect();
            }
        }

   这个地方调用了ParallelNioSender的sendMessage()方法后，还调用了keepalive(),这个方法去除NioSender的集合中不在存活状态的NioSender.最后归还NioSender.

 

  看下ParallelNioSender的sendMessage()方法，主要是先拿到需要传播的NioSender数组，之后一个个的去传送。

 如果NioSender数组在超时的时间限制内没有发送完成的话，就会抛出异常.主要是在doLoop()方法里面执行的，这里如果出现异常的话，还有NioSender重连等机制试图恢复NioSender的功能

 

 

 

看tomcat的设计，关于资源池的设计，基本上有两个集合，一个是已经使用的集合，一个是未使用的集合，当未使用集合还有资源的话，现在未使用的集合，当没有的话，在小于最大限度的情况下，可以重新new一个，当一个资源使用完毕的话，放入到未使用的集合可以重新利用。有些时候，一个资源是否可以放入未使用集合是有考虑的，比如他的持续的服务时间是否过长，如果过长就不会放在，因为这个资源过老的话，可能利用的资源不能得到释放，最坏的情况是资源被耗尽，如内存泄漏等，这样还不如重新new一个新的，这些都是我们设计资源池需要考虑的