Jetty : Embedded Server 启动流程 - 2

摘要：前言本文接着启动流程往下讲，上回说到调用方法，开始接收和处理请求，默认使用的子类，所以我们重点来看看的具体实现类层次，我们上文提到过，这里有两个新面孔抽象实现类上文讲过方法最终会调用方法，使用模板方法模式在方法中实现了的基本流程子类方法，

本文接着 Jetty : Embedded Server 启动流程 - 1往下讲，上回说到 Server.start 调用 Connector.start 方法，开始接收和处理请求，Server 默认使用 Connector 的子类 SelectChannelConnector，所以我们重点来看看 SelectChannelConnector 的具体实现

AbstractLifeCycle
    AggregateLifeCycle
        AbstractConnector
            AbstractNIOConnector
                SelectChannelConnector

AbstractLifeCycle，AggregateLifeCycle 我们上文提到过，这里有两个新面孔：

AbstractConnector：Connector 抽象实现类

AbstractNIOConnector

上文讲过 Server.start 方法最终会调用 Connector.start 方法，AbstractConnector 使用《模板方法》模式在 doStart 方法中实现了 start 的基本流程

@Override
protected void doStart() throws Exception {
    if (_server == null) {
        throw new IllegalStateException("No server");
    }
    
    // 子类 override open 方法，打开 server socket
    open();

    // 如果没有指定 ThreadPool，默认使用 Server 的 ThreadPool
    if (_threadPool == null) {
        _threadPool = _server.getThreadPool();
        addBean(_threadPool, false);
    }
    ...
    synchronized (this) {
        _acceptorThreads = new Thread[getAcceptors()];
        for (int i = 0; i < _acceptorThreads.length; i++) {
            // 启动 acceptor 线程监听客户端连接
            if (!_threadPool.dispatch(new Acceptor(i))) {
                throw new IllegalStateException("!accepting");
            }
            ...
        }
    }
    ...
}

几个关键点：

子类通过 override（覆盖）open 方法初始化 server socket

通过 setAcceptors 方法可以设置 acceptor 线程数量

如果没有特殊指定，acceptor 线程 和 请求处理线程在（Server）同一个线程池里头

Acceptor 类（实现了 Runnable 接口）实现了具体的 accept 方法

Acceptor 类是 AbstractConnector 类的内部类，它在 Run 方法里头调用 AbstractConnector.accept 方法接收客户端连接

private class Acceptor implements Runnable {
    public void run() {
        Thread current = Thread.currentThread();
        ...
        accept(_acceptor);
        ...
    }
}

accept 方法是个抽象方法，由 AbstractConnector 的子类提供具体实现

我们重点关注和启动流程相关的三个方法:open，doStart，accept 以及一个类 SelectorManager

上文提到 AbstractConnector 在 start 方法中调用 open 方法，子类 override open 方法打开 server socket

    public void open() throws IOException {
        synchronized(this) {
            if (_acceptChannel == null) {
                // Create a new server socket
                _acceptChannel = ServerSocketChannel.open();
                // Set to blocking mode，阻塞接收连接请求
                _acceptChannel.configureBlocking(true);

                // Bind the server socket to the local host and port
                _acceptChannel.socket().setReuseAddress(getReuseAddress());
                InetSocketAddress addr = getHost()==null ?
                    new InetSocketAddress(getPort()) :
                    new InetSocketAddress(getHost(), getPort());
                _acceptChannel.socket().bind(addr,getAcceptQueueSize());

                _localPort=_acceptChannel.socket().getLocalPort();
                if (_localPort<=0)
                    throw new IOException("Server channel not bound");

                addBean(_acceptChannel);
            }
        }
    }

@Override
public void accept(int acceptorID) throws IOException {
    ServerSocketChannel server;
    synchronized(this) {
        server = _acceptChannel;
    }

    if (server!=null && server.isOpen() && _manager.isStarted()) {
        // 获取客户端 SocketChannel
        SocketChannel channel = server.accept();
        // 设置非阻塞模式（NIO）
        channel.configureBlocking(false);
        Socket socket = channel.socket();
        configure(socket);
        // 将 channel 注册到 SelectorManager（见下文）
        _manager.register(channel);
    }
}

通过 server.accept 获取到客户端 SocketChannel，并将它注册到 _manager（SelectorManager），这个 _manager 是啥？

The Selector Manager manages and number of SelectSets to allow NIO Scheduling to scale to large numbers of connections
注意 SelectChannelConnector 在构造方法里将 _manager 作为 managed bean 添加到 bean registry 里，这样在 SelectChannelConnector 启动(start 方法被调用)的时候 SelectManager 也会跟着启动（参考上文）

public class SelectChannelConnector extends AbstractNIOConnector {
    private final SelectorManager _manager = new ConnectorSelectorManager();

    public SelectChannelConnector() {
        _manager.setMaxIdleTime(getMaxIdleTime());
        addBean(_manager, true);
        ...
    }
}

SelectManager doStart 方法，这里只保留方法主要逻辑

@Override
protected void doStart() throws Exception {
    _selectSet = new SelectSet[_selectSets];
    for (int i = 0; i < _select.length; i++) {
        _selectSet[i] = new SelectSet(i);
    }
    super.doStart();
    // start a thread to select
    for (int i = 0; i < getSelectSets(); i++) {
        final int id = i;
        // 提交 select runnable 到线程池
        boolean selecting = dispatch(new Runnalbe() {
            public void run() {
                ...
                LOG.debug("Starting {} on {}", Thread.currentThread(), this);
                while (isRunning()) {
                    try {
                        // 对注册的 channel 进行多路选择（select）
                        set.doSelect();
                    } catch (...) {
                        ...
                    }
                }
            }
        });
    }
}

doStart 方法启动 _selectSet 个线程监听 channel select 事件，我们回过头来看 SelectManager 的 register 方法

public void register(SocketChannel acceptChannel) {
    int s=_set++;
    if (s<0) {
        s=-s;
    }
    s=s%_selectSets;
    SelectSet set=_selectSet[s];
    set.addChange(acceptChannel);
    set.wakeup();
}

acceptChannel 被均匀分配（addChange）给 SelectSet

到目前为止我们总结一下：

Server 启动 N 个 Accept 线程接收客户端连接

Server 启动 N 个 Select 线程对 SocketChannel 进行多路IO选择，每个线程执行 SelectSet 的 doSelect 方法

Server 接收到客户端连接后将 SocketChannel 注册到 SelectorManager

SelectorManager 将注册的 SocketChannel 均匀分配给各个 SelectSet，同时唤醒 Select 线程迎接新的客户端请求