Mina 抽象Polling连接器（AbstractPollingIoConnector）

Mina 连接器接口定义及抽象实现（IoConnector ）：http://donald-draper.iteye.com/blog/2378936
引言：
IoConnector接口给Ioservice增加了连接功能，可以连接服务端。连接操作，首先检查连接器状态，本地地址与远程地址是否为空已经与传输元数据地址类型是否匹配，如果连接器Iohandler为null，创建一个对会话操作事件不处理的IoHandler，最后将实际连接操作委托给connect0，待子类实现。

/**
 * A base class for implementing client transport using a polling strategy. The
 * underlying sockets will be checked in an active loop and woke up when an
 * socket needed to be processed. This class handle the logic behind binding,
 * connecting and disposing the client sockets. A {@link Executor} will be used
 * for running client connection, and an {@link AbstractPollingIoProcessor} will
 * be used for processing connected client I/O operations like reading, writing
 * and closing.
 *
 * All the low level methods for binding, connecting, closing need to be
 * provided by the subclassing implementation.
 * 
 * @see NioSocketConnector for a example of implementation
 * @param <H> The type of IoHandler
 * @param <S> The type of IoSession
 * 
 * @author [url=http://mina.apache.org]Apache MINA Project[/url]
 */
public abstract class AbstractPollingIoConnector<S extends AbstractIoSession, H> extends AbstractIoConnector {
   //连接请求队列
    private final Queue<ConnectionRequest> connectQueue = new ConcurrentLinkedQueue<>();
    //连接关闭或取消队列
    private final Queue<ConnectionRequest> cancelQueue = new ConcurrentLinkedQueue<>();
    private final IoProcessor<S> processor;//Io处理器
    private final boolean createdProcessor;
    private final ServiceOperationFuture disposalFuture = new ServiceOperationFuture();//关闭结果
    private volatile boolean selectable;
    /** The connector thread 连接线程引用*/
    private final AtomicReference<Connector> connectorRef = new AtomicReference<>();
}

从上面可以看出，抽象拉取连接器内部比较重要的几个变量为连接请求队列connectQueue，连接请求取消队列cancelQueue，
Io处理器，连接线程引用connectorRef。
再来看构造:

 /**
  * Constructor for {@link AbstractPollingIoConnector}. You need to provide a
  * default session configuration, a class of {@link IoProcessor} which will
  * be instantiated in a {@link SimpleIoProcessorPool} for better scaling in
  * multiprocessor systems. The default pool size will be used.
  * 构造抽象拉取连接器，需要提供默认的会话配置，和一个IO处理器，用于在SimpleIoProcessorPool
  中创建实例，默认线程池size，适用与多处理器系统。
  * @see SimpleIoProcessorPool
  * 
  * @param sessionConfig
  *            the default configuration for the managed {@link IoSession}
  * @param processorClass
  *            a {@link Class} of {@link IoProcessor} for the associated
  *            {@link IoSession} type.
  */
 protected AbstractPollingIoConnector(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass) {
     this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass), true);
 }

 /**
  * Constructor for {@link AbstractPollingIoConnector}. You need to provide a
  * default session configuration, a class of {@link IoProcessor} which will
  * be instantiated in a {@link SimpleIoProcessorPool} for using multiple
  * thread for better scaling in multiprocessor systems.
  * 与上一个方法不同的时，多个个处理器线程池size参数
  * @see SimpleIoProcessorPool
  * 
  * @param sessionConfig
  *            the default configuration for the managed {@link IoSession}
  * @param processorClass
  *            a {@link Class} of {@link IoProcessor} for the associated
  *            {@link IoSession} type.
  * @param processorCount
  *            the amount of processor to instantiate for the pool
  */
 protected AbstractPollingIoConnector(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass,
         int processorCount) {
     this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass, processorCount), true);
 }

 /**
  * Constructor for {@link AbstractPollingIoConnector}. You need to provide a
  * default session configuration, a default {@link Executor} will be created
  * using {@link Executors#newCachedThreadPool()}.
  * 需要提供默认会话配置，与上两个方法不同是，传输的Io处理器实例，service共享Io处理器，
  一个默认的线程池为Executors#newCachedThreadPool
  * @see AbstractIoService#AbstractIoService(IoSessionConfig, Executor)
  * 
  * @param sessionConfig
  *            the default configuration for the managed {@link IoSession}
  * @param processor
  *            the {@link IoProcessor} for processing the {@link IoSession}
  *            of this transport, triggering events to the bound
  *            {@link IoHandler} and processing the chains of
  *            {@link IoFilter}
  */
 protected AbstractPollingIoConnector(IoSessionConfig sessionConfig, IoProcessor<S> processor) {
     this(sessionConfig, null, processor, false);
 }

 /**
  * Constructor for {@link AbstractPollingIoConnector}. You need to provide a
  * default session configuration and an {@link Executor} for handling I/O
  * events. If null {@link Executor} is provided, a default one will be
  * created using {@link Executors#newCachedThreadPool()}.
  * 与上面不同的是添加IO事件执行器参数Executor
  * @see AbstractIoService#AbstractIoService(IoSessionConfig, Executor)
  * 
  * @param sessionConfig
  *            the default configuration for the managed {@link IoSession}
  * @param executor
  *            the {@link Executor} used for handling asynchronous execution
  *            of I/O events. Can be <code>null</code>.
  * @param processor
  *            the {@link IoProcessor} for processing the {@link IoSession}
  *            of this transport, triggering events to the bound
  *            {@link IoHandler} and processing the chains of
  *            {@link IoFilter}
  */
 protected AbstractPollingIoConnector(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor) {
     this(sessionConfig, executor, processor, false);
 }

上面所有的构造方法，都是通过AbstractPollingIoConnector(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor,
boolean createdProcessor)来实现，我们来看这个方法：

/**
  * Constructor for {@link AbstractPollingIoAcceptor}. You need to provide a
  * default session configuration and an {@link Executor} for handling I/O
  * events. If null {@link Executor} is provided, a default one will be
  * created using {@link Executors#newCachedThreadPool()}.
  * 
  * @see AbstractIoService#AbstractIoService(IoSessionConfig, Executor)
  * 
  * @param sessionConfig
  *            the default configuration for the managed {@link IoSession}
  * @param executor
  *            the {@link Executor} used for handling asynchronous execution
  *            of I/O events. Can be <code>null</code>.
  * @param processor
  *            the {@link IoProcessor} for processing the {@link IoSession}
  *            of this transport, triggering events to the bound
  *            {@link IoHandler} and processing the chains of
  *            {@link IoFilter}
  * @param createdProcessor
  *            tagging the processor as automatically created, so it will be
  *            automatically disposed
  */
 private AbstractPollingIoConnector(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor,
         boolean createdProcessor) {
    //初始化会话配置，IO事件执行器和IO处理器
     super(sessionConfig, executor);
     if (processor == null) {
         throw new IllegalArgumentException("processor");
     }
     this.processor = processor;
     this.createdProcessor = createdProcessor;
     try {
         init();
         selectable = true;
     } catch (RuntimeException e) {
         throw e;
     } catch (Exception e) {
         throw new RuntimeIoException("Failed to initialize.", e);
     } finally {
         if (!selectable) {
             try {
                 destroy();
             } catch (Exception e) {
                 ExceptionMonitor.getInstance().exceptionCaught(e);
             }
         }
     }
 }  
 /**
  * Initialize the polling system, will be called at construction time.
  * 初始化poll系统，在构造是调用
  * @throws Exception
  *             any exception thrown by the underlying system calls
  */
 protected abstract void init() throws Exception;

从上面可以看出，拉取连接器构造主要初始化会话配置，IO事件执行器和IO处理器。
再来看其他方法的定义：

/**
 * Destroy the polling system, will be called when this {@link IoConnector}
 * implementation will be disposed.
 * 销毁连接器
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract void destroy() throws Exception;

/**
 * Create a new client socket handle from a local {@link SocketAddress}
 * 从本地socket地址创建一个的客户端handle（SocketChannel）
 * @param localAddress
 *            the socket address for binding the new client socket
 * @return a new client socket handle
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract H newHandle(SocketAddress localAddress) throws Exception;

/**
 * Connect a newly created client socket handle to a remote
 * {@link SocketAddress}. This operation is non-blocking, so at end of the
 * call the socket can be still in connection process.
 * 根据远端socket和本地socket地址创建一个客户端SocketChannel。此操作为非阻塞模式，
 在方法结束，可能能在连接中
 * @param handle the client socket handle
 * @param remoteAddress the remote address where to connect
 * @return <tt>true</tt> if a connection was established, <tt>false</tt> if
 *         this client socket is in non-blocking mode and the connection
 *         operation is in progress
 * @throws Exception If the connect failed
 */
protected abstract boolean connect(H handle, SocketAddress remoteAddress) throws Exception;

/**
 * Finish the connection process of a client socket after it was marked as
 * ready to process by the {@link #select(int)} call. The socket will be
 * connected or reported as connection failed.
 * 在选择操作标记客户端SocketChannel连接完毕调用
 * @param handle
 *            the client socket handle to finish to connect
 * @return true if the socket is connected
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract boolean finishConnect(H handle) throws Exception;

/**
 * Create a new {@link IoSession} from a connected socket client handle.
 * Will assign the created {@link IoSession} to the given
 * {@link IoProcessor} for managing future I/O events.
 * 根据SocketChannel和IO处理器创建一个IO会话
 * @param processor
 *            the processor in charge of this session
 * @param handle
 *            the newly connected client socket handle
 * @return a new {@link IoSession}
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract S newSession(IoProcessor<S> processor, H handle) throws Exception;

/**
 * Close a client socket.
 * 关闭SocketChannel客户端
 * @param handle
 *            the client socket
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract void close(H handle) throws Exception;

/**
 * Interrupt the {@link #select(int)} method. Used when the poll set need to
 * be modified.
 中断选择操作
 */
protected abstract void wakeup();

/**
 * Check for connected sockets, interrupt when at least a connection is
 * processed (connected or failed to connect). All the client socket
 * descriptors processed need to be returned by {@link #selectedHandles()}
 * 检查客户端来接服务器是否成功，当至少一个连接操作完成时中断（消除中断位），所有客户端描述符可以通过
#selectedHandles方法返回
 * @param timeout The timeout for the select() method
 * @return The number of socket having received some data
 * @throws Exception any exception thrown by the underlying systems calls
 */
protected abstract int select(int timeout) throws Exception;

/**
 * {@link Iterator} for the set of client sockets found connected or failed
 * to connect during the last {@link #select(int)} call.
 * 获取在上次选择操作调用后，连接成功或失败的客户端handler集
 * @return the list of client socket handles to process
 */
protected abstract Iterator<H> selectedHandles();

/**
 * {@link Iterator} for all the client sockets polled for connection.
 * 连接器所有poll的客户端
 * @return the list of client sockets currently polled for connection
 */
protected abstract Iterator<H> allHandles();

/**
 * Register a new client socket for connection, add it to connection polling
 * 注册一个连接客户端，添加到连接polling系统
 * @param handle
 *            client socket handle
 * @param request
 *            the associated {@link ConnectionRequest}
 * @throws Exception
 *             any exception thrown by the underlying systems calls
 */
protected abstract void register(H handle, ConnectionRequest request) throws Exception;

/**
 * get the {@link ConnectionRequest} for a given client socket handle
 * 获取连接客户端SocketChannel的连接请求
 * @param handle
 *            the socket client handle
 * @return the connection request if the socket is connecting otherwise
 *         <code>null</code>
 */
protected abstract ConnectionRequest getConnectionRequest(H handle);

上面这些方法都是抽象的，子类实现，我们简单看一下，以便理解连接操作，我们来看连接操作:

*
 * {@inheritDoc}
 */
@Override
@SuppressWarnings("unchecked")
protected final ConnectFuture connect0(SocketAddress remoteAddress, SocketAddress localAddress,
        IoSessionInitializer<? extends ConnectFuture> sessionInitializer) {
    H handle = null;
    boolean success = false;
    try {
        //根据本地socket地址创建SocketChannel，连接远端socket地址
        handle = newHandle(localAddress);
        if (connect(handle, remoteAddress)) {
            ConnectFuture future = new DefaultConnectFuture();
	    //根据IO处理器和SocketChannel构建Io会话
            S session = newSession(processor, handle);
            initSession(session, future, sessionInitializer);
            // Forward the remaining process to the IoProcessor.
	    //将会话添加到会话关联的IO处理器中
            session.getProcessor().add(session);
            success = true;
            return future;
        }
        success = true;
    } catch (Exception e) {
        return DefaultConnectFuture.newFailedFuture(e);
    } finally {
        if (!success && handle != null) {
            try {
                close(handle);
            } catch (Exception e) {
                ExceptionMonitor.getInstance().exceptionCaught(e);
            }
        }
    }
    //根据SocketChannel和会话初始化sessionInitializer构建连接请求，添加到连接请求队列。
    ConnectionRequest request = new ConnectionRequest(handle, sessionInitializer);
    connectQueue.add(request);
    //启动连接器线程
    startupWorker();
    wakeup();
    return request;
}

//启动连接器线程

private void startupWorker() {
    if (!selectable) {
        connectQueue.clear();
        cancelQueue.clear();
    }
    Connector connector = connectorRef.get();
    if (connector == null) {
        connector = new Connector();
        if (connectorRef.compareAndSet(null, connector)) {
            executeWorker(connector);
        }
    }
}

从上面可以看出连接操作，首先根据本地socket地址创建SocketChannel，连接远端socket地址，根据IO处理器和SocketChannel构建Io会话，将会话添加到会话关联的IO处理器中，
根据SocketChannel和会话初始化sessionInitializer构建连接请求，添加到连接请求队列，
最后启动连接器线程。
我们来看连接器线程的定义：
//Connector

private class Connector implements Runnable {
    /**
     * {@inheritDoc}
     */
    @Override
    public void run() {
        assert connectorRef.get() == this;
        int nHandles = 0;
        while (selectable) {
            try {
                // the timeout for select shall be smaller of the connect
                // timeout or 1 second...
		//选择超时时间
                int timeout = (int) Math.min(getConnectTimeoutMillis(), 1000L);
		//执行选择操作
                int selected = select(timeout);
		//从连接请求队列poll连接请求，注册连接请求SocketChannel连接事件到选择器
                nHandles += registerNew();

                // get a chance to get out of the connector loop, if we
                // don't have any more handles
		//如果没有任何连接请求SocketChannel需要处理
                if (nHandles == 0) {
		    //置空连接器连接线程引用
                    connectorRef.set(null);
		    //清空连接请求队列
                    if (connectQueue.isEmpty()) {
                        assert connectorRef.get() != this;
                        break;
                    }
                    if (!connectorRef.compareAndSet(null, this)) {
                        assert connectorRef.get() != this;
                        break;
                    }
                    assert connectorRef.get() == this;
                }
                //如果有连接请求已经连接完成，即触发SocketChannel兴趣连接事件
                if (selected > 0) {
		    //处理连接事件就绪的连接请求
                    nHandles -= processConnections(selectedHandles());
                }
                //处理超时的连接请求
                processTimedOutSessions(allHandles());
                //处理取消连接的连接请求
                nHandles -= cancelKeys();
            } catch (ClosedSelectorException cse) {
                // If the selector has been closed, we can exit the loop
                ExceptionMonitor.getInstance().exceptionCaught(cse);
                break;
            } catch (Exception e) {
                ExceptionMonitor.getInstance().exceptionCaught(e);

                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e1) {
                    ExceptionMonitor.getInstance().exceptionCaught(e1);
                }
            }
        }

        if (selectable && isDisposing()) {
            selectable = false;
            try {
                if (createdProcessor) {
		    //释放Io处理器
                    processor.dispose();
                }
            } finally {
                try {
                    synchronized (disposalLock) {
                        if (isDisposing()) {
			    //销毁连接器
                            destroy();
                        }
                    }
                } catch (Exception e) {
                    ExceptionMonitor.getInstance().exceptionCaught(e);
                } finally {
		    //设置连接器已关闭
                    disposalFuture.setDone();
                }
            }
        }
    }
    //注册连接请求SocketChannel连接事件到选择器
    private int registerNew() {
        int nHandles = 0;
        for (;;) {
	    //从连接队列获取连接请求
            ConnectionRequest req = connectQueue.poll();
            if (req == null) {
                break;
            }
            //从连接请求获取客户端SocketChannel
            H handle = req.handle;
            try {
	        //注册SocketChannel连接事件到选择器
                register(handle, req);
                nHandles++;
            } catch (Exception e) {
                req.setException(e);
                try {
                    close(handle);
                } catch (Exception e2) {
                    ExceptionMonitor.getInstance().exceptionCaught(e2);
                }
            }
        }
        return nHandles;
    }
    //处理取消连接的连接请求
    private int cancelKeys() {
        int nHandles = 0;
        //遍历取消连接请求队列，关闭连接请求关联的SocketChannel
        for (;;) {
            ConnectionRequest req = cancelQueue.poll();
            if (req == null) {
                break;
            }
            H handle = req.handle;
            try {
                close(handle);
            } catch (Exception e) {
                ExceptionMonitor.getInstance().exceptionCaught(e);
            } finally {
                nHandles++;
            }
        }

        if (nHandles > 0) {
            wakeup();
        }

        return nHandles;
    }

    /**
     * Process the incoming connections, creating a new session for each valid
     * connection.
     处理连接事件就绪的连接请求
     */
    private int processConnections(Iterator<H> handlers) {
        int nHandles = 0;
        // Loop on each connection request
	//遍历连接请求队列的SocketChannel
        while (handlers.hasNext()) {
            H handle = handlers.next();
            handlers.remove();
            ConnectionRequest connectionRequest = getConnectionRequest(handle);
            if (connectionRequest == null) {
                continue;
            }
            boolean success = false;
            try {
	        //调用finishConnect完成SocketChannel连接后续工作
                if (finishConnect(handle)) {
		    //根据Io处理器和SocketChannel创建会话
                    S session = newSession(processor, handle);
		    //初始化会话
                    initSession(session, connectionRequest, connectionRequest.getSessionInitializer());
                    // Forward the remaining process to the IoProcessor.
		    //添加会话到会话关联的IO处理器
                    session.getProcessor().add(session);
                    nHandles++;
                }
                success = true;
            } catch (Exception e) {
	        //设置连接请求异常
                connectionRequest.setException(e);
            } finally {
                if (!success) {
                    // The connection failed, we have to cancel it.
		    //如果处理失败，则添加连接请求到取消队列
                    cancelQueue.offer(connectionRequest);
                }
            }
        }
        return nHandles;
    }
    //处理连接超时的连接请求
    private void processTimedOutSessions(Iterator<H> handles) {
        long currentTime = System.currentTimeMillis();
	//遍历所有连接请求的SocketChannel，设置连接结果超时异常，添加到连接请求到取消队列
        while (handles.hasNext()) {
            H handle = handles.next();
            ConnectionRequest connectionRequest = getConnectionRequest(handle);
            if ((connectionRequest != null) && (currentTime >= connectionRequest.deadline)) {
                connectionRequest.setException(new ConnectException("Connection timed out."));
                cancelQueue.offer(connectionRequest);
            }
        }
    }
}

从上面可以看出，连接器线程首先计算选择超时时间，执行超时选择操作，注册连接请求SocketChannel连接事件到选择器；如果没有任何连接请求SocketChannel需要处理，置空连接器连接线程引用，清空连接请求队列，如果有连接请求已经连接完成，即触发SocketChannel兴趣连接事件，处理连接事件就绪的连接请求，这个过程首先调用finishConnect完成SocketChannel连接后续工作，根据Io处理器和SocketChannel创建会话，初始化会话，添加会话到会话关联的IO处理器；然后处理连接超时的连接请求，即设置连接结果超时异常，添加到连接请求到取消队列；处理取消连接的连接请求，即关闭连接请求关联的SocketChannel。

总结：
     抽象拉取连接器内部有一个连接请求队列connectQueue，连接请求取消队列cancelQueue，Io处理器和连接线程引用connectorRef。拉取连接器构造主要初始化会话配置，IO事件执行器和IO处理器。连接操作，首先根据本地socket地址创建SocketChannel，连接远端socket地址，根据IO处理器和SocketChannel构建Io会话，将会话添加到会话关联的IO处理器中，根据SocketChannel和会话初始化sessionInitializer构建连接请求，添加到连接请求队列，最后启动连接器线程。
     连接器线程首先计算选择超时时间，执行超时选择操作，注册连接请求SocketChannel连接事件到选择器；如果没有任何连接请求SocketChannel需要处理，置空连接器连接线程引用，清空连接请求队列，如果有连接请求已经连接完成，即触发SocketChannel兴趣连接事件，处理连接事件就绪的连接请求，这个过程首先调用finishConnect完成SocketChannel连接后续工作，根据Io处理器和SocketChannel创建会话，初始化会话，添加会话到会话关联的IO处理器；然后处理连接超时的连接请求，即设置连接结果超时异常，添加到连接请求到取消队列；处理取消连接的连接请求，即关闭连接请求关联的SocketChannel。