欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识！
您现在的位置是： 首页

Mina 抽象Polling连接器（AbstractPollingIoConnector）

程序员文章站 2022-03-11 09:29:52
...
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。

相关标签： mina

上一篇： AE CS6怎么制作怀旧老照片效果? AE图片褪色的技巧

下一篇： Mina IoHandler接口定义

推荐阅读