Mina 框架源码解析-NioSocketAcceptor启动过程

Mina 框架源码解析-NioSocketAcceptor启动过程

在HeartBeatServer 中，有一句启动Server端的代码

// 创建一个非阻塞的server端的Socket
acceptor = new NioSocketAcceptor();

 下面分析这个 new NioSocketAcceptor的过程：

 

1，NioSocketAcceptor 类头声明

public final class NioSocketAcceptor
        extends AbstractPollingIoAcceptor<NioSession, ServerSocketChannel>
        implements SocketAcceptor {

 继承了AbstractPollingIoAcceptor 类，实现SocketAcceptor 接口

 

2，看AbstractPollingIoAcceptor 类，

类头声明

public abstract class AbstractPollingIoAcceptor<S extends AbstractIoSession, H>
        extends AbstractIoAcceptor {

继承了AbstractIoAcceptor 类，AbstractIoAcceptor 类说明请看第3点

在AbstractPollingIoAcceptor 类加载的时候初始化了这些

 private final Queue<AcceptorOperationFuture> registerQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();

 private final Queue<AcceptorOperationFuture> cancelQueue = new ConcurrentLinkedQueue<AcceptorOperationFuture>();

 private final Map<SocketAddress, H> boundHandles = Collections.synchronizedMap(new HashMap<SocketAddress, H>());

 private final ServiceOperationFuture disposalFuture = new ServiceOperationFuture();

    /** The thread responsible of accepting incoming requests */ 
 private AtomicReference<Acceptor> acceptorRef = new AtomicReference<Acceptor>();

 

3，AbstractIoAcceptor 类头声明

public abstract class AbstractIoAcceptor 
        extends AbstractIoService implements IoAcceptor {

 继承AbstractIoService 类，AbstractIoService类实现IoService接口

public abstract class AbstractIoService implements IoService {

AbstractIoAcceptor 类在加载时初始化这些

    private final List<SocketAddress> defaultLocalAddresses =
        new ArrayList<SocketAddress>();
    private final List<SocketAddress> unmodifiableDefaultLocalAddresses =
        Collections.unmodifiableList(defaultLocalAddresses);
    private final Set<SocketAddress> boundAddresses =
        new HashSet<SocketAddress>();

    private boolean disconnectOnUnbind = true;

    /**
     * 当bind或unbind操作的时候需要获得的锁对象
     * The lock object which is acquired while bind or unbind operation is performed.
     * Acquire this lock in your property setters which shouldn't be changed while
     * the service is bound.
     */
    protected final Object bindLock = new Object();

 

 

4，SocketAcceptor 接口，AbstractPollingIoAcceptor 构造函数等先不看，那只不过是实现一种规范而已。

 

5，回到NioSocketAcceptor 类，进入构造方法

    public NioSocketAcceptor() {
    	/** 使用默认的Socket配置 
    	 * 调用父类的构造器  ,将默认配置传入到父类中，第二个参数NioProcessor.class 是动态类加载，会放到线程池上运行
    	 * 在父类AbstractPollingIoAcceptor的构造方法中调用init()接口，在子类中会调用init（）接口的实现，因而初始化了selector
    	 * protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig,
         *     Class<? extends IoProcessor<S>> processorClass) {
         * 	this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass),true);
    	 *	}
    	 * */
        super(new DefaultSocketSessionConfig(), NioProcessor.class);//标记2
      
        ((DefaultSocketSessionConfig) getSessionConfig()).init(this);
    }

 super方法（标记2），调用父类AbstractPollingIoAcceptor 构造方法，初始化了DefaultSocketSessionConfig 类，并和NioProcessor .class 传递给上一级构造方法,构造方法如下

protected AbstractPollingIoAcceptor(IoSessionConfig sessionConfig,
            Class<? extends IoProcessor<S>> processorClass) {
    	//初始化selector，捕获一些异常，将处理器类加载进来
        this(sessionConfig, null, new SimpleIoProcessorPool<S>(processorClass),
                true);//标记3
    }

 this使用（标记3），接收到下级传来的DefaultSocketSessionConfig和NioProcessor， 并用SimpleIoProcessorPool初始化，（SimpleIoProcessorPool将在下一节介绍），再次传递给AbstractPollingIoAcceptor 类的构造方法

 private AbstractPollingIoAcceptor(IoSessionConfig sessionConfig,
            Executor executor, IoProcessor<S> processor,
            boolean createdProcessor) {
        super(sessionConfig, executor);//标记4

        if (processor == null) {
            throw new IllegalArgumentException("processor");
        }

        this.processor = processor;//new SimpleIoProcessorPool()
        this.createdProcessor = createdProcessor;//是否创建新的处理池

        try {
            // Initialize the selector
            init();//初始化selector，因此在此处是抽象类，所以最终执行的是实体类的init()，也就是继承了此抽象类的实体类NioSocketAcceptor
            
            // 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);
                }
            }
        }
    }

 super方法（标记4），调用父类AbstractIoAcceptor 的构造方法

   protected AbstractIoAcceptor(IoSessionConfig sessionConfig, Executor executor) {
        super(sessionConfig, executor);//标记5
        defaultLocalAddresses.add(null);
    }

 此处super方法（标记5），调用父类AbstractIoService 的构造方法，线程池等在此处初始化

/**
     * 构造器，必须提供一个session配置和一个线程执行器，如果线程执行器为空，则会自动创建一个新的线程池
     * Constructor for {@link AbstractIoService}. 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()}.
     * 
     * @param sessionConfig
     *            the default configuration for the managed {@link IoSession}
     * @param executor
     *            the {@link Executor} used for handling execution of I/O
     *            events. Can be <code>null</code>.
     */
    protected AbstractIoService(IoSessionConfig sessionConfig, Executor executor) {
        if (sessionConfig == null) {
            throw new IllegalArgumentException("sessionConfig");
        }

        if (getTransportMetadata() == null) {
            throw new IllegalArgumentException("TransportMetadata");
        }

        if (!getTransportMetadata().getSessionConfigType().isAssignableFrom(
                sessionConfig.getClass())) {
            throw new IllegalArgumentException("sessionConfig type: "
                    + sessionConfig.getClass() + " (expected: "
                    + getTransportMetadata().getSessionConfigType() + ")");
        }

        // Create the listeners, and add a first listener : a activation listener
        // for this service, which will give information on the service state.
        listeners = new IoServiceListenerSupport(this);
        listeners.add(serviceActivationListener);

        // Stores the given session configuration
        this.sessionConfig = sessionConfig;

        // Make JVM load the exception monitor before some transports
        // change the thread context class loader.
        ExceptionMonitor.getInstance();

        if (executor == null) {
            this.executor = Executors.newCachedThreadPool();
            createdExecutor = true;
        } else {
            this.executor = executor;
            createdExecutor = false;
        }

        threadName = getClass().getSimpleName() + '-' + id.incrementAndGet();
    }

 初始化Listener监听机制，就完成了new NioSocketAcceptor()的过程，Listener监听机制将在以后讲解，下一节是

Mina SimpleIoProcessor 的实现

评论

1 楼 xiucaiyao 2014-06-05