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

Mina 抽象polling监听器

程序员文章站 2022-03-11 09:37:16
...
Mina Io监听器接口定义及抽象实现:http://donald-draper.iteye.com/blog/2378315
引言:
    IoAcceptor与IoService不同的是,添加了监听连接请求和地址绑定功能。抽象Io监听器AbstractIoAcceptor绑定地址首先要检查绑定的socket地址与传输元数据的地址类型是否相同,相同则通过bindInternal完成实际的绑定,然后通知Service监听器,Service已激活fireServiceActivated。解绑地址方法,主要是委托unbind0方法完成实际解绑工作,清空绑定地址集合boundAddresses,触发Service监听器无效事件fireServiceDeactivated。
/**
 * A base class for implementing 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,
 * accepting and disposing the server sockets. An {@link Executor} will be used
 * for running client accepting and an {@link AbstractPollingIoProcessor} will
 * be used for processing client I/O operations like reading, writing and
 * closing.
 * AbstractPollingIoAcceptor为传输层拉取策略的基本实现。底层socket将在一个循环中检查,连接请求事件,当一个以个socket连接请求时,唤醒监听器线程,处理连接请求。此类处理地址绑定,接收连接请求,释放server socket。一个执行器将会运行接收客户端连接,AbstractPollingIoProcessor将用于处理客户端的IO操作事件。
 * All the low level methods for binding, accepting, closing need to be provided
 * by the subclassing implementation.
 * 
 * @see NioSocketAcceptor 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 AbstractPollingIoAcceptor<S extends AbstractIoSession, H> extends AbstractIoAcceptor {
    /** A lock used to protect the selector to be waked up before it's created */
    //信号量用于在选择器被创建前,保护选择器被唤醒
    private final Semaphore lock = new Semaphore(1);
    private final IoProcessor<S> processor;//Io处理器
    private final boolean createdProcessor;//Io处理器是否创建
    //地址绑定请求队列,地址绑定时,添加绑定请求AcceptorOperationFuture到队列
    private final Queue<AcceptorOperationFuture> registerQueue = new ConcurrentLinkedQueue<>();
    //解绑队列,解绑地址时,添加地址解绑请求AcceptorOperationFuture到队列
    private final Queue<AcceptorOperationFuture> cancelQueue = new ConcurrentLinkedQueue<>();
    //监听器绑定的socket地址,与ServerSocketChannel映射关系
    //绑定地址后,添加地址与ServerSocketChannel映射关系到boundHandles
    private final Map<SocketAddress, H> boundHandles = Collections.synchronizedMap(new HashMap<SocketAddress, H>());
    private final ServiceOperationFuture disposalFuture = new ServiceOperationFuture();//Service关闭结果
    /** A flag set when the acceptor has been created and initialized */
    private volatile boolean selectable;//当acceptor创建和初始化设置
    /** The thread responsible of accepting incoming requests 监听连接请求线程Acceptor*/
    private AtomicReference<Acceptor> acceptorRef = new AtomicReference<>();
    protected boolean reuseAddress = false;//地址是否重用
    /**
     * Define the number of socket that can wait to be accepted. Default
     * to 50 (as in the SocketServer default).
     SocketServer默认可以接收的连接数。
     */
    protected int backlog = 50;
 }

来看AbstractPollingIoAcceptor的构造:
    
/**
     * Constructor for {@link AbstractPollingIoAcceptor}. 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.
     * 构造AbstractPollingIoAcceptor,需要提供默认的会话配置,在IO简单处理器线程池当中,执行的Io处理器类型。
     默认的线程池大小将会被使用
     * @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 AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass) {
        this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass), true, null);
    }

    /**
     * Constructor for {@link AbstractPollingIoAcceptor}. 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.
     * 与上面的方法不同的时,指定Io处理器线程池大小,即Io处理器的个数。
     * @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 AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass,
            int processorCount) {
        this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass, processorCount), true, null);
    }

    /**
     * Constructor for {@link AbstractPollingIoAcceptor}. 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.
     *与上面不同的时,多个一个选择器提供者参数
     * @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
     * @param selectorProvider The SelectorProvider to use
     */
    protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Class<? extends IoProcessor<S>> processorClass,
            int processorCount, SelectorProvider selectorProvider ) {
        this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass, processorCount, selectorProvider), true, selectorProvider);
    }

    /**
     * Constructor for {@link AbstractPollingIoAcceptor}. You need to provide a default
     * session configuration, a default {@link Executor} will be created using
     * {@link Executors#newCachedThreadPool()}.
     * 此方法的Io处理器执行器,为 Executors#newCachedThreadPool
     * @see AbstractIoService
     * 
     * @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 AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, IoProcessor<S> processor) {
        this(sessionConfig, null, processor, false, null);
    }

    /**
     * Constructor for {@link AbstractPollingIoAcceptor}. You need to provide a
     * default session configuration and an {@link Executor} for handling I/O
     * events. If a null {@link Executor} is provided, a default one will be
     * created using {@link Executors#newCachedThreadPool()}.
     * 如果线程池为空,默认为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}
     */
    protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor) {
        this(sessionConfig, executor, processor, false, null);
    }

    /**
     * Constructor for {@link AbstractPollingIoAcceptor}. You need to provide a
     * default session configuration and an {@link Executor} for handling I/O
     * events. If a null {@link Executor} is provided, a default one will be
     * created using {@link Executors#newCachedThreadPool()}.
     * 这个就是最终的构造方法了。
     * @see #AbstractIoService(IoSessionConfig, Executor)
     * 
     * @param sessionConfig 默认会话配置
     *            the default configuration for the managed {@link IoSession}
     * @param executor 异步IO事件执行器
     *            the {@link Executor} used for handling asynchronous execution
     *            of I/O events. Can be <code>null</code>.
     * @param processor 会话处理器器,触发IoHander和过滤器的相关事件
     *            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 AbstractPollingIoAcceptor(IoSessionConfig sessionConfig, Executor executor, IoProcessor<S> processor,
            boolean createdProcessor, SelectorProvider selectorProvider) {
        super(sessionConfig, executor);
        if (processor == null) {
            throw new IllegalArgumentException("processor");
        }
        this.processor = processor;
        this.createdProcessor = createdProcessor;
        try {
            // Initialize the selector
            init(selectorProvider);初始化选择器
            // The selector is now ready, we can switch the
            // flag to true so that incoming connection can be accepted
            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.
     * 
     * @param selectorProvider The Selector Provider that will be used by this polling acceptor
     * @throws Exception any exception thrown by the underlying system calls
     */
    protected abstract void init(SelectorProvider selectorProvider) throws Exception;

从上面来看,构造AbstractPollingIoAcceptor,主要是初始化会话配置,Io处理器类型,IO异步事件执行器为空的话默认为CachedThreadPool,然后初始化选择器。
再来看其他方法:

  
 /**
     * Initialize the polling system, will be called at construction time.
     * @throws Exception any exception thrown by the underlying system calls
     初始化拉取系统,在构造是调用
     */
    protected abstract void init() throws Exception;

  
    /**
     * Destroy the polling system, will be called when this {@link IoAcceptor}
     * implementation will be disposed.
     销毁拉取系统,在IO监听器销毁时,调用
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract void destroy() throws Exception;

    /**
     * Check for acceptable connections, interrupt when at least a server is ready for accepting.
     * All the ready server socket descriptors need to be returned by {@link #selectedHandles()}
     检查已接收的连接,当至少有一个准备好接收时,中断。所有的就绪serversocket描述符需要通过
     #selectedHandles方法返回
     * @return The number of sockets having got incoming client 接收的客户端连接数
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract int select() throws Exception;

    /**
     * Interrupt the {@link #select()} method. Used when the poll set need to be modified.
     唤醒,选择操作。当poll及修改时
     */
    protected abstract void wakeup();

    /**
     * {@link Iterator} for the set of server sockets found with acceptable incoming connections
     *  during the last {@link #select()} call.
     返回在上一次选择操作中,服务端接收连接数
     * @return the list of server handles ready
     */
    protected abstract Iterator<H> selectedHandles();
    
    /**
     * Open a server socket for a given local address.
     打开一个给定socket地址的serversocket
     * @param localAddress the associated local address
     * @return the opened server socket
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract H open(SocketAddress localAddress) throws Exception;

    /**
     * Get the local address associated with a given server socket
     获取给定serversocket的本地地址
     * @param handle the server socket
     * @return the local {@link SocketAddress} associated with this handle
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract SocketAddress localAddress(H handle) throws Exception;

    /**
     * Accept a client connection for a server socket and return a new {@link IoSession}
     * associated with the given {@link IoProcessor}
     serversocket接收一个客户端连接请求,返回一个关联指定IO处理器的会话。
     * @param processor the {@link IoProcessor} to associate with the {@link IoSession}
     * @param handle the server handle
     * @return the created {@link IoSession}
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract S accept(IoProcessor<S> processor, H handle) throws Exception;

    /**
     * Close a server socket. 关闭一个serversocket
     * @param handle the server socket
     * @throws Exception any exception thrown by the underlying systems calls
     */
    protected abstract void close(H handle) throws Exception;

再来绑定socket地址:
/**
 * {@inheritDoc}
 */
@Override
protected final Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses) throws Exception {
    // Create a bind request as a Future operation. When the selector
    // have handled the registration, it will signal this future.
    //创建绑定操作结果。当选择器处理注册时,将会通知操作结果。
    AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);
    // adds the Registration request to the queue for the Workers
    // to handle 注册请求到注册队列,以便工作线程处理
    registerQueue.add(request);

    // creates the Acceptor instance and has the local
    // executor kick it off.
    //创建监听器实例,本地执行器将会执行实例
    startupAcceptor();

    // As we just started the acceptor, we have to unblock the select()
    // in order to process the bind request we just have added to the
    // registerQueue.
    //由于我们刚启动监听器,不得不unblock选择操作,处理刚刚添加到注册队列的绑定请求
    try {
        lock.acquire();

        wakeup();
    } finally {
        lock.release();
    }

    // Now, we wait until this request is completed.
    request.awaitUninterruptibly();

    if (request.getException() != null) {
        throw request.getException();
    }
    // Update the local addresses.
    // setLocalAddresses() shouldn't be called from the worker thread
    // because of deadlock.
    Set<SocketAddress> newLocalAddresses = new HashSet<>();
    for (H handle : boundHandles.values()) {
        newLocalAddresses.add(localAddress(handle));
    }
    return newLocalAddresses;
}

地址绑定有一点要关注:
// creates the Acceptor instance and has the local 
// executor kick it off.                           
//创建监听器实例,本地执行器将会执行实例           
startupAcceptor();  


 /**
 * This method is called by the doBind() and doUnbind()
 * methods.  If the acceptor is null, the acceptor object will
 * be created and kicked off by the executor.  If the acceptor
 * object is null, probably already created and this class
 * is now working, then nothing will happen and the method
 * will just return.
 */
private void startupAcceptor() throws InterruptedException {
    // If the acceptor is not ready, clear the queues
    // TODO : they should already be clean : do we have to do that ?
    if (!selectable) {
        //如果acceptor没有准备好,则清空注册队列和取消队列
        registerQueue.clear();
        cancelQueue.clear();
    }
    // start the acceptor if not already started,
    //获取当前正在运行的acceptor,没有这创建一个,有执行器执行。
    Acceptor acceptor = acceptorRef.get();
    if (acceptor == null) {
        lock.acquire();
        acceptor = new Acceptor();

        if (acceptorRef.compareAndSet(null, acceptor)) {
            executeWorker(acceptor);
        } else {
            lock.release();
        }
    }
}

我们再来看监听器Acceptor的定义:
 /**
   * This class is called by the startupAcceptor() method and is
   * placed into a NamePreservingRunnable class.
  * It's a thread accepting incoming connections from clients.
  * The loop is stopped when all the bound handlers are unbound.
  Acceptor监听器,在startupAcceptor方法,创建被包装成NamePreservingRunnable线程
  执行。Acceptor是一个线程,接收来自客户端的连接。当所有绑定的Handler解绑时,循环停止。
  */
 private class Acceptor implements Runnable {
     /**
      * {@inheritDoc}
      */
     @Override
     public void run() {
         assert acceptorRef.get() == this;
         int nHandles = 0;
         // Release the lock
         lock.release();
         //当监听器已经准备好
         while (selectable) {
             try {
                 // Process the bound sockets to this acceptor.
                 // this actually sets the selector to OP_ACCEPT,
                 // and binds to the port on which this class will
                 // listen on. We do that before the select because 
                 // the registerQueue containing the new handler is
                 // already updated at this point.
		 //根据地址绑定队列中的绑定请求,打开一个ServerSocketChannle,
		 //注册接收事件OP_ACCEPT到选择器
                 nHandles += registerHandles();

                 // Detect if we have some keys ready to be processed
                 // The select() will be woke up if some new connection
                 // have occurred, or if the selector has been explicitly
                 // woke up
		 //探测是否存在选择key就绪待处理,及一个新的连接请求发生,或者
		 //选择操作被显示唤醒,执行选择操作
                 int selected = select();

                 // Now, if the number of registred handles is 0, we can
                 // quit the loop: we don't have any socket listening
                 // for incoming connection.
                 if (nHandles == 0) {
		     //如果绑定地址为空,则置空acceptorRef
                     acceptorRef.set(null);

                     if (registerQueue.isEmpty() && cancelQueue.isEmpty()) {
                         assert acceptorRef.get() != this;
                         break;
                     }

                     if (!acceptorRef.compareAndSet(null, this)) {
                         assert acceptorRef.get() != this;
                         break;
                     }

                     assert acceptorRef.get() == this;
                 }

                 if (selected > 0) {
                     // We have some connection request, let's process
                     // them here.
		     //有接收连接事件发生,则处理连接请求
                     processHandles(selectedHandles());
                 }

                 // check to see if any cancellation request has been made.
		 //unbind解绑队列的解绑地址请求
                 nHandles -= unregisterHandles();
             } 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);
                 }
             }
         }

         // Cleanup all the processors, and shutdown the acceptor.
         if (selectable && isDisposing()) {
	    //如果acceptor正在关闭,则关闭关联processor
             selectable = false;
             try {
                 if (createdProcessor) {
                     processor.dispose();
                 }
             } finally {
                 try {
                     synchronized (disposalLock) {
                         if (isDisposing()) {
                             destroy();
                         }
                     }
                 } catch (Exception e) {
                     ExceptionMonitor.getInstance().exceptionCaught(e);
                 } finally {
                     disposalFuture.setDone();
                 }
             }
         }
     }
     /**
      * This method will process new sessions for the Worker class.  All
      * keys that have had their status updates as per the Selector.selectedKeys()
      * method will be processed here.  Only keys that are ready to accept
      * connections are handled here.
      * <p/>
      * Session objects are created by making new instances of SocketSessionImpl
      * and passing the session object to the SocketIoProcessor class.
      */
     @SuppressWarnings("unchecked")
     private void processHandles(Iterator<H> handles) throws Exception {
        //遍历接收事件就绪的ServerSocketChannel
         while (handles.hasNext()) {
             H handle = handles.next();
             handles.remove();

             // Associates a new created connection to a processor,
             // and get back a session
	     //ServerSocketChannel创建一个关联processor的会话
             S session = accept(processor, handle);
             if (session == null) {
                 continue;
             }
	     //初始化会话
             initSession(session, null, null);
             // add the session to the SocketIoProcessor
	     //添加会话到会话关联io处理器
             session.getProcessor().add(session);
         }
     }
     /**
      * Sets up the socket communications.  Sets items such as:
     绑定socket地址,打开一个ServerSocketChannle,注册接收事件OP_ACCEPT到选择器
      * <p/>
      * Blocking
      * Reuse address
      * Receive buffer size
      * Bind to listen port
      * Registers OP_ACCEPT for selector
      */
     private int registerHandles() {
         for (;;) {
             // The register queue contains the list of services to manage
             // in this acceptor.
	     //从socket地址绑定请求队列,poll一个地址绑定请求
             AcceptorOperationFuture future = registerQueue.poll();

             if (future == null) {
                 return 0;
             }

             // We create a temporary map to store the bound handles,
             // as we may have to remove them all if there is an exception
             // during the sockets opening.
             Map<SocketAddress, H> newHandles = new ConcurrentHashMap<>();
	     //获取需要绑定的地址
             List<SocketAddress> localAddresses = future.getLocalAddresses();
             try {
                 // Process all the addresses
                 for (SocketAddress a : localAddresses) {
                     H handle = open(a);//根据Socket地址,打开一个ServerSocketChannle
                     newHandles.put(localAddress(handle), handle);
                 }

                 // Everything went ok, we can now update the map storing
                 // all the bound sockets.
		 //将绑定地址与ServerSocketChannle映射管理添加到地址绑定映射集合
                 boundHandles.putAll(newHandles);

                 // and notify.
                 future.setDone();
                 
                 return newHandles.size();
             } catch (Exception e) {
                 // We store the exception in the future
                 future.setException(e);
             } finally {
                 // Roll back if failed to bind all addresses.
                 if (future.getException() != null) {
		     //如果绑定地址有异常,则关闭打开的ServerSocketChannle
                     for (H handle : newHandles.values()) {
                         try {
                             close(handle);
                         } catch (Exception e) {
                             ExceptionMonitor.getInstance().exceptionCaught(e);
                         }
                     }

                     // Wake up the selector to be sure we will process the newly bound handle
                     // and not block forever in the select()
		     //唤醒选择器处理新创建的ServerSocketChannle,不在阻塞在选择操作
                     wakeup();
                 }
             }
         }
     }

     /**
      * This method just checks to see if anything has been placed into the
      * cancellation queue.  The only thing that should be in the cancelQueue
      * is CancellationRequest objects and the only place this happens is in
      * the doUnbind() method.
      unregisterHandles方法,检查是否有地址解绑请求。如果有解绑请求CancellationRequest,
      解绑请求在doUnbind方法中产生
      */
     private int unregisterHandles() {
         int cancelledHandles = 0;
         for (;;) {
	 //从解绑队列中poll解绑请求
             AcceptorOperationFuture future = cancelQueue.poll();
             if (future == null) {
                 break;
             }
             // close the channels
	     //获取ServerSocketChannle绑定的地址
             for (SocketAddress a : future.getLocalAddresses()) {
	        //从绑定socket地址与ServerSocketChannle映射map中,移除绑定的socket地址
                 H handle = boundHandles.remove(a);
                 if (handle == null) {
                     continue;
                 }
                 try {
		     //关闭ServerSocketChannle
                     close(handle);
		     //唤醒,触发ServerSocketChannle线程death
                     wakeup(); // wake up again to trigger thread death
                 } catch (Exception e) {
                     ExceptionMonitor.getInstance().exceptionCaught(e);
                 } finally {
                     cancelledHandles++;
                 }
             }
             //解绑成功
             future.setDone();
         }

         return cancelledHandles;
     }
}

从Acceptor的源码可以看出,如果监听器AbstractPollingIoAcceptor已经初始化,首先
根据地址绑定队列中的绑定请求,打开一个ServerSocketChannle,注册接收事件OP_ACCEPT到选择器,并将绑定地址与ServerSocketChannle映射管理添加到地址绑定映射集合;执行选择操作,如果实际绑定地址为空,则置空acceptorRef;如果接收连接事件发生,则处理连接请求,遍历接收事件就绪的ServerSocketChannel,ServerSocketChannel创建一个关联processor的会话,初始化会话,添加会话到会话关联io处理器;检查是否有地址解绑请求,如果有解绑请求CancellationRequest,从绑定socket地址与ServerSocketChannle映射map中,移除绑定的socket地址,关闭ServerSocketChannle;最后检查监听器是否正在关闭,如果acceptor正在关闭,则关闭关联processor。
回到地址绑定方法:
protected final Set<SocketAddress> bindInternal(List<? extends SocketAddress> localAddresses) throws Exception {
    // Create a bind request as a Future operation. When the selector
    // have handled the registration, it will signal this future.
    //创建绑定操作结果。当选择器处理注册时,将会通知操作结果。
    AcceptorOperationFuture request = new AcceptorOperationFuture(localAddresses);
    // adds the Registration request to the queue for the Workers
    // to handle 注册请求到注册队列,以便工作线程处理
    registerQueue.add(request);

    // creates the Acceptor instance and has the local
    // executor kick it off.
    //创建监听器实例,本地执行器将会执行实例
    startupAcceptor();

    // As we just started the acceptor, we have to unblock the select()
    // in order to process the bind request we just have added to the
    // registerQueue.
    //由于我们刚启动监听器,不得不unblock选择操作,处理刚刚添加到注册队列的绑定请求
    try {
        lock.acquire();

        wakeup();
    } finally {
        lock.release();
    }

    // Now, we wait until this request is completed.
    request.awaitUninterruptibly();

    if (request.getException() != null) {
        throw request.getException();
    }
    // Update the local addresses.
    // setLocalAddresses() shouldn't be called from the worker thread
    // because of deadlock.
    Set<SocketAddress> newLocalAddresses = new HashSet<>();
    for (H handle : boundHandles.values()) {
        newLocalAddresses.add(localAddress(handle));
    }
    return newLocalAddresses;
}

 /**
 * This method is called by the doBind() and doUnbind()
 * methods.  If the acceptor is null, the acceptor object will
 * be created and kicked off by the executor.  If the acceptor
 * object is null, probably already created and this class
 * is now working, then nothing will happen and the method
 * will just return.
 */
private void startupAcceptor() throws InterruptedException {
    // If the acceptor is not ready, clear the queues
    // TODO : they should already be clean : do we have to do that ?
    if (!selectable) {
        //如果acceptor没有准备好,则清空注册队列和取消队列
        registerQueue.clear();
        cancelQueue.clear();
    }
    // start the acceptor if not already started,
    //获取当前正在运行的acceptor,没有这创建一个,有执行器执行。
    Acceptor acceptor = acceptorRef.get();
    if (acceptor == null) {
        lock.acquire();
        acceptor = new Acceptor();

        if (acceptorRef.compareAndSet(null, acceptor)) {
            executeWorker(acceptor);
        } else {
            lock.release();
        }
    }
}


从上面来看,地址绑定过程为,创建绑定操作结果,注册绑定请求到注册地址绑定请求队列,
创建监听器Acceptor实例并执行。
再来看地址解绑等操作:
  
 /**
     * {@inheritDoc}
     */
    @Override
    protected final void unbind0(List<? extends SocketAddress> localAddresses) throws Exception {
        //根据解绑地址创建AcceptorOperationFuture,添加到解绑队列,启动Acceptor线程,完成实际解绑工作。
        AcceptorOperationFuture future = new AcceptorOperationFuture(localAddresses);
        cancelQueue.add(future);
        startupAcceptor();
        wakeup();
        future.awaitUninterruptibly();
        if (future.getException() != null) {
            throw future.getException();
        }
    }
    /**
     * {@inheritDoc}
     */
    @Override
    protected void dispose0() throws Exception {
        unbind();//解绑地址
        startupAcceptor();//启动Acceptor线程,完成实际清理工作。
        wakeup();
    }
    /**
     * {@inheritDoc}
     默认不支持根据远端地址和本地地址创建会话
     */
    @Override
    public final IoSession newSession(SocketAddress remoteAddress, SocketAddress localAddress) {
        throw new UnsupportedOperationException();
    }
    /**
     * @return the backLog
     */
    public int getBacklog() {
        return backlog;
    }
    /**
     * Sets the Backlog parameter
     * 
     * @param backlog
     *            the backlog variable
     */
    public void setBacklog(int backlog) {
        synchronized (bindLock) {
            if (isActive()) {
                throw new IllegalStateException("backlog can't be set while the acceptor is bound.");
            }

            this.backlog = backlog;
        }
    }
    /**
     * @return the flag that sets the reuseAddress information
     */
    public boolean isReuseAddress() {
        return reuseAddress;
    }
    /**
     * Set the Reuse Address flag
     * 
     * @param reuseAddress
     *            The flag to set
     */
    public void setReuseAddress(boolean reuseAddress) {
        synchronized (bindLock) {
            if (isActive()) {
                throw new IllegalStateException("backlog can't be set while the acceptor is bound.");
            }
            this.reuseAddress = reuseAddress;
        }
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public SocketSessionConfig getSessionConfig() {
        return (SocketSessionConfig)sessionConfig;
    }


总结:
AbstractPollingIoAcceptor主要变量为Io处理器processor,地址绑定请求队列registerQueue,地址解绑请求队列cancelQueue,监听器绑定的socket地址,与ServerSocketChannel映射关系boundHandles-Map,监听工作线程Acceptor引用acceptorRef。
构造AbstractPollingIoAcceptor,主要是初始化会话配置,Io处理器类型,IO异步事件执行器为空的话默认为CachedThreadPool,然后初始化选择器。地址绑定过程为,创建绑定操作结果,注册绑定请求到注册地址绑定请求队列,创建监听器Acceptor实例并执行。Acceptor主要功能为,地址绑定,监听连接请求,解绑地址,实际工作逻辑为:如果监听器AbstractPollingIoAcceptor已经初始化,首先根据地址绑定队列中的绑定请求,打开一个ServerSocketChannle,注册接收事件OP_ACCEPT到选择器,并将绑定地址与ServerSocketChannle映射管理添加到地址绑定映射集合;执行选择操作,如果实际绑定地址为空,则置空acceptorRef;如果接收连接事件发生,则处理连接请求,遍历接收事件就绪的ServerSocketChannel,ServerSocketChannel创建一个关联processor的会话,初始化会话,添加会话到会话关联io处理器;检查是否有地址解绑请求,如果有解绑请求CancellationRequest,从绑定socket地址与ServerSocketChannle映射map中,移除绑定的socket地址,关闭ServerSocketChannle;最后检查监听器是否正在关闭,如果acceptor正在关闭,则关闭关联processor。Acceptor和AbstractPollingIoAcceptor的关系,与AbstractPollingIoProcessor和Processor的关系很像。地址解绑过程,首先根据解绑地址创建AcceptorOperationFuture,添加到解绑队列,启动Acceptor线程,完成实际解绑工作。
AbstractPollingIoAcceptor所有的工作(地址绑定,接收连接,创建会话,添加会话到IO处理器,解绑地址,释放监听器资源)都是在Acceptor线程里完成。


附:
//SimpleIoProcessorPool,简单的IO处理器线程池,这一步,可以做个了解。
简单的IO处理器线程池SimpleIoProcessorPool,将一个会话的相关事件在多个Io处理器执行。当前的transport内部实现用SimpleIoProcessorPool,在多处理器环境下,具有良好的性能,因此不需要直接使用SimpleIoProcessorPool,除非在同个虚拟机中运行多个IoService。
Io处理器线程池,主要是创建size个IO处理器线程处理会话和过滤器相关事件。如果在同一虚拟机中运行多个Io服务,你需要共享处理器线程池。为了达到这个效果,你需要构造
SimpleIoProcessorPool实例,在创建IO服务时,作为参数传入。会话与IO处理器线程池,主要通过会话的处理器属性关联,获取会话关联的处理器,如果会话属性没有存储关联处理器,则从处理器线程池中获取一个,添加会话属性中。SimpleIoProcessorPool的其他方法发送写请求,刷新会话,添加移除会话等操作都是通过会话关联的处理器线程池中的处理器。简单的说,SimpleIoProcessorPool使用与再同一个虚拟机下启动多个IO服务的情况,简单处理器线程池,就是一个处理器集合,添加会话时,从会话属性中,获会话关联的处理器,如果会话属性没有存储关联处理器,则从处理器线程池中获取一个,添加会话属性中。
/**
 * An {@link IoProcessor} pool that distributes {@link IoSession}s into one or more
 * {@link IoProcessor}s. Most current transport implementations use this pool internally
 * to perform better in a multi-core environment, and therefore, you won't need to 
 * use this pool directly unless you are running multiple {@link IoService}s in the
 * same JVM.
简单的IO处理器线程池SimpleIoProcessorPool,将一个会话的相关事件在多个Io处理器执行。
当前的transport内部实现用SimpleIoProcessorPool,在多处理器环境下,具有良好的性能,
因此不需要直接使用SimpleIoProcessorPool,除非在同个虚拟机中运行多个IoService。
 * <p>
 * If you are running multiple {@link IoService}s, you could want to share the pool
 * among all services.  To do so, you can create a new {@link SimpleIoProcessorPool}
 * instance by yourself and provide the pool as a constructor parameter when you
 * create the services.
 如果在同一虚拟机中运行多个Io服务,你需要共享处理器线程池。为了达到这个效果,你需要构造
SimpleIoProcessorPool实例,在创建IO服务时,作为参数传入。
 * <p>
 * This pool uses Java reflection API to create multiple {@link IoProcessor} instances.
 * It tries to instantiate the processor in the following order:
Io处理器线程池默认通过反射创建io处理器实例,有一下三中形式
 * [list=1]
 * [*]A public constructor with one {@link ExecutorService} parameter.

 * [*]A public constructor with one {@link Executor} parameter.

 * [*]A public default constructor

 * [/list]
 * The following is an example for the NIO socket transport:
 * <pre><code>
 下面是创建TCP服务端的一个实例
 * // Create a shared pool.
 * SimpleIoProcessorPool<NioSession> pool = 
 *         new SimpleIoProcessorPool<NioSession>(NioProcessor.class, 16);
 * 
 * // Create two services that share the same pool.
 * SocketAcceptor acceptor = new NioSocketAcceptor(pool);
 * SocketConnector connector = new NioSocketConnector(pool);
 * 
 * ...
 * 
 * // Release related resources.
 * connector.dispose();
 * acceptor.dispose();
 * pool.dispose();
 * </code></pre>
 * 
 * @author [url=http://mina.apache.org]Apache MINA Project[/url]
 * 
 * @param <S> the type of the {@link IoSession} to be managed by the specified
 *            {@link IoProcessor}.
 */
public class SimpleIoProcessorPool<S extends AbstractIoSession> implements IoProcessor<S> {
    /** A logger for this class */
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleIoProcessorPool.class);
    /** The default pool size, when no size is provided.默认处理器实例数为运行时环境可以利用处理器数量+1 */
    private static final int DEFAULT_SIZE = Runtime.getRuntime().availableProcessors() + 1;
    /** A key used to store the processor pool in the session's Attributes */
    private static final AttributeKey PROCESSOR = new AttributeKey(SimpleIoProcessorPool.class, "processor");
    /** The pool table */
    private final IoProcessor<S>[] pool;//io处理器集会话的相关事件。
    /** The contained  which is passed to the IoProcessor when they are created */
    private final Executor executor;//io异步事件执行器
    /** A flag set to true if we had to create an executor */
    private final boolean createdExecutor;
    /** A lock to protect the disposal against concurrent calls */
    private final Object disposalLock = new Object();
    /** A flg set to true if the IoProcessor in the pool are being disposed */
    private volatile boolean disposing;
    /** A flag set to true if all the IoProcessor contained in the pool have been disposed */
    private volatile boolean disposed;

    /**
     * Creates a new instance of SimpleIoProcessorPool with a default
     * size of NbCPUs +1.
     *
     * @param processorType The type of IoProcessor to use
     */
    public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType) {
        this(processorType, null, DEFAULT_SIZE, null);
    }
    /**
     * Creates a new instance of SimpleIoProcessorPool with a defined
     * number of IoProcessors in the pool
     *
     * @param processorType The type of IoProcessor to use
     * @param size The number of IoProcessor in the pool
     */
    public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, int size) {
        this(processorType, null, size, null);
    }
    /**
     * Creates a new instance of SimpleIoProcessorPool with a defined
     * number of IoProcessors in the pool
     *
     * @param processorType The type of IoProcessor to use
     * @param size The number of IoProcessor in the pool
     * @param selectorProvider The SelectorProvider to use
     */
    public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, int size, SelectorProvider selectorProvider) {
        this(processorType, null, size, selectorProvider);
    }
    /**
     * Creates a new instance of SimpleIoProcessorPool with an executor
     *
     * @param processorType The type of IoProcessor to use
     * @param executor The {@link Executor}
     */
    public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor) {
        this(processorType, executor, DEFAULT_SIZE, null);
    }
    /**
     * Creates a new instance of SimpleIoProcessorPool with an executor
     *
     * @param processorType The type of IoProcessor to use
     * @param executor The {@link Executor}
     * @param size The number of IoProcessor in the pool
     * @param selectorProvider The SelectorProvider to used
     */
    @SuppressWarnings("unchecked")
    public SimpleIoProcessorPool(Class<? extends IoProcessor<S>> processorType, Executor executor, int size, 
            SelectorProvider selectorProvider) {
        if (processorType == null) {
            throw new IllegalArgumentException("processorType");
        }
        if (size <= 0) {
            throw new IllegalArgumentException("size: " + size + " (expected: positive integer)");
        }
        // Create the executor if none is provided
        createdExecutor = executor == null;
        if (createdExecutor) {
            this.executor = Executors.newCachedThreadPool();
            // Set a default reject handler
            ((ThreadPoolExecutor) this.executor).setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        } else {
            this.executor = executor;
        }
        pool = new IoProcessor[size];//初始化IO处理器集
        boolean success = false;
        Constructor<? extends IoProcessor<S>> processorConstructor = null;
        boolean usesExecutorArg = true;
        try {
            // We create at least one processor
            try {
                try {
		    //创建处理器
                    processorConstructor = processorType.getConstructor(ExecutorService.class);
                    pool[0] = processorConstructor.newInstance(this.executor);
                } catch (NoSuchMethodException e1) {
                    // To the next step...
                    try {
		        //如果处理器没有相关构造方法,重新构造Io处理器
                        if(selectorProvider==null) {
                            processorConstructor = processorType.getConstructor(Executor.class);
                            pool[0] = processorConstructor.newInstance(this.executor);
                        } else {
                            processorConstructor = processorType.getConstructor(Executor.class, SelectorProvider.class);
                            pool[0] = processorConstructor.newInstance(this.executor,selectorProvider);
                        }
                    } catch (NoSuchMethodException e2) {
                        // To the next step...
                        try {
                            processorConstructor = processorType.getConstructor();
                            usesExecutorArg = false;
                            pool[0] = processorConstructor.newInstance();
                        } catch (NoSuchMethodException e3) {
                            // To the next step...
                        }
                    }
                }
            } catch (RuntimeException re) {
                LOGGER.error("Cannot create an IoProcessor :{}", re.getMessage());
                throw re;
            } catch (Exception e) {
                String msg = "Failed to create a new instance of " + processorType.getName() + ":" + e.getMessage();
                LOGGER.error(msg, e);
                throw new RuntimeIoException(msg, e);
            }

            if (processorConstructor == null) {
                // Raise an exception if no proper constructor is found.
                String msg = String.valueOf(processorType) + " must have a public constructor with one "
                        + ExecutorService.class.getSimpleName() + " parameter, a public constructor with one "
                        + Executor.class.getSimpleName() + " parameter or a public default constructor.";
                LOGGER.error(msg);
                throw new IllegalArgumentException(msg);
            }

            // Constructor found now use it for all subsequent instantiations
            for (int i = 1; i < pool.length; i++) {
                try {
                    if (usesExecutorArg) {
                        if(selectorProvider==null) {
                            pool[i] = processorConstructor.newInstance(this.executor);
                        } else {
                            pool[i] = processorConstructor.newInstance(this.executor, selectorProvider);
                        }
                    } else {
                        pool[i] = processorConstructor.newInstance();
                    }
                } catch (Exception e) {
                    // Won't happen because it has been done previously
                }
            }

            success = true;
        } finally {
            if (!success) {
                dispose();
            }
        }
    }
   //从上来看,构造Io处理器线程池,主要是创建size个IO处理器线程处理会话和过滤器相关事件
    /**
     * {@inheritDoc}
     */
    @Override
    public final void add(S session) {
        getProcessor(session).add(session);
    }
     /**
     * Find the processor associated to a session. If it hasen't be stored into
     * the session's attributes, pick a new processor and stores it.
     获取会话关联的处理器,如果会话属性没有存储关联处理器,则从处理器线程池中获取一个,添加
     会话属性中。
     */
    @SuppressWarnings("unchecked")
    private IoProcessor<S> getProcessor(S session) {
        IoProcessor<S> processor = (IoProcessor<S>) session.getAttribute(PROCESSOR);
        if (processor == null) {
            if (disposed || disposing) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }
            processor = pool[Math.abs((int) session.getId()) % pool.length];

            if (processor == null) {
                throw new IllegalStateException("A disposed processor cannot be accessed.");
            }
            session.setAttributeIfAbsent(PROCESSOR, processor);
        }

        return processor;
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public final void flush(S session) {
        getProcessor(session).flush(session);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public final void write(S session, WriteRequest writeRequest) {
        getProcessor(session).write(session, writeRequest);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public final void remove(S session) {
        getProcessor(session).remove(session);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public final void updateTrafficControl(S session) {
        getProcessor(session).updateTrafficControl(session);
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isDisposed() {
        return disposed;
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public boolean isDisposing() {
        return disposing;
    }
    /**
     * {@inheritDoc}
     */
    @Override
    public final void dispose() {
        if (disposed) {
            return;
        }
        synchronized (disposalLock) {
            if (!disposing) {
                disposing = true;
                //遍历处理器集,释放处理器资源
                for (IoProcessor<S> ioProcessor : pool) {
                    if (ioProcessor == null) {
                        // Special case if the pool has not been initialized properly
                        continue;
                    }

                    if (ioProcessor.isDisposing()) {
                        continue;
                    }

                    try {
                        ioProcessor.dispose();
                    } catch (Exception e) {
                        LOGGER.warn("Failed to dispose the {} IoProcessor.", ioProcessor.getClass().getSimpleName(), e);
                    }
                }

                if (createdExecutor) {
                    ((ExecutorService) executor).shutdown();
                }
            }

            Arrays.fill(pool, null);
            disposed = true;
        }
    }
}
相关标签: mina