Jetty源码导读二：接受请求过程

Jetty的请求入口

ServerConnector.java 的 accepted 方法（ServerSocketChannel#accept 后的处理逻辑）。

Jetty的请求流程

一个请求的流程：

1.Acceptor 监听连接请求，当有连接请求到达时就接受连接，一个连接对应一个 Channel，Acceptor 将 Channel 交给 ManagedSelector 来处理。
2.ManagedSelector 把 Channel 注册到 Selector 上，并创建一个 EndPoint 和 Connection 跟这个 Channel 绑定，接着就不断地检测 I/O 事件。
3.I/O 事件到了就调用 EndPoint 的方法拿到一个 Runnable，并扔给线程池执行。
4.线程池中调度某个线程执行 Runnable。
5.Runnable 执行时，调用回调函数，这个回调函数是 Connection 注册到 EndPoint 中的。
6.回调函数内部实现，其实就是调用 EndPoint 的接口方法来读数据。
7.Connection 解析读到的数据，生成请求对象并交给 Handler 组件去处理。

SelectorManager

Jetty 的 Selector 由 SelectorManager 类管理，而被管理的 Selector 叫作 ManagedSelector。（这里SelectorManager 的具体实现类是 ServerConnector的内部类 ServerConnectorManager。） SelectorManager 内部有一个 ManagedSelector 数组，真正干活的是 ManagedSelector。主要做两个事情：

选择一个 ManagedSelector 来处理 Channel
提交一个任务 Accept 给 ManagedSelector

// SelectorManager.java public void accept(SelectableChannel channel, Object attachment) { // 选择一个 ManagedSelector 来处理 Channel final ManagedSelector selector = chooseSelector(); // 提交一个任务 Accept 给 ManagedSelector selector.submit(selector.new Accept(channel, attachment)); }

ManagedSelector

ManagedSelector 在处理这个任务Accept 主要做了两步：

第一步，调用 Selector 的 register 方法把 Channel 注册到 Selector 上，拿到一个 SelectionKey。

// ManagedSelector$Accept.java 内部类 @Override public void update(Selector selector){ try{ // 把 Channel 注册到 Selector 上，拿到一个 SelectionKey key = channel.register(selector, 0, attachment); execute(this); // 执行当前Runnable, 跳转 run() } catch (Throwable x) { IO.close(channel); _selectorManager.onAcceptFailed(channel, x); LOG.debug(x); } } @Override public void run(){ try{ // 创建一个 EndPoint 和 Connection，并跟这个 SelectionKey（Channel）绑在一起 createEndPoint(channel, key); _selectorManager.onAccepted(channel); } catch (Throwable x){ LOG.debug(x); failed(x); } }
第二步，创建一个 EndPoint 和 Connection，并跟这个 SelectionKey（Channel）绑在一起：

// ManagedSelector.java private void createEndPoint(SelectableChannel channel, SelectionKey selectionKey) throws IOException{ //1. 创建 Endpoint EndPoint endPoint = _selectorManager.newEndPoint(channel, this, selectionKey); //2. 创建 Connection Connection connection = _selectorManager.newConnection(channel, endPoint, selectionKey.attachment()); //3. 把 Endpoint、Connection 和 SelectionKey 绑在一起 endPoint.setConnection(connection); selectionKey.attach(endPoint); endPoint.onOpen(); endPointOpened(endPoint); // 将Connection 注入到 EndPoint, 跳转 Connection 的 onOpen 方法 (内部转 fillInterested) _selectorManager.connectionOpened(connection); }

如上，HttpConnection (Connection 的具体实现类之一) 并不会主动向 EndPoint 读取数据，而是向在 EndPoint 中注册一堆回调方法：

1// HttpConnection.java
2public void onOpen(){
3    super.onOpen();
4    if (isRequestBufferEmpty())
5        fillInterested();
6    else
7        getExecutor().execute(this);
8}
9public void fillInterested(){
10    // 告诉 EndPoint，数据到了你就调我这些回调方法 _readCallback 吧，有点异步 I/O 的感觉，也就是说 Jetty 在应用层面模拟了异步 I/O 模型。
11    getEndPoint().fillInterested(_readCallback);
12}
13
14

ManagedSelector 的 EatWhatYouKill

这时候，ManagedSelector 启动时候的任务 EatWhatYouKill 的无限循环，监测到SelectorProducer的 processSelected 方法选择出一个 Runnable 则，会自动执行。这个 Runnable 从上步骤来看，则是调用EndPoint 的 onSelected 方法返回一个 Runnable，然后把这个 Runnable 直接执行或扔给线程池执行。

1// ManagedSelector.java
2_selectorManager.execute(_strategy::produce);
3
4
5
6
7// ManagedSelector$SelectorProducer.java
8public Runnable produce() {
9    while (true) {
10        // 查看key绑定的 EndPoint 是否有可执行的回调函数 （读 或 写 或 NULL）
11        // 获得后递交全局线程池执行
12        // 或 ManagedSelector$Acceptor 执行接受客户端连接的函数, 返回 NULL
13        Runnable task = processSelected();
14        if (task != null) {
15            return task;
16        }
17        // 执行 ManagedSelector$Acceptor 注册 SelectionKey.OP_ACCEPT 接受事件
18        processUpdates();
19        // 执行 Endpoint 的 update 方法，切换注册 读、写事件
20        updateKeys();
21        // 唤醒select, 获取就绪的key
22        if (!select()) {
23            return null;
24        }
25    }
26}

具体里面就是调用 EndPoint的 onSelected方法

1// ManagedSelector$SelectorProducer.java
2...
3if (attachment instanceof Selectable){
4    // Try to produce a task
5    Runnable task = ((Selectable)attachment).onSelected();
6    if (task != null)
7        return task;
8}
9
10

EndPoint (ChannelEndPoint)

（这里EndPoint 的具体实现类是 ChannelEndPoint）

这里怎么选择出这个 Runnable 呢？？

1// ChannelEndPoint.java
2
3    public Runnable onSelected() {
4
5        ....
6
7        boolean fillable = (readyOps & SelectionKey.OP_READ) != 0;
8
9        boolean flushable = (readyOps & SelectionKey.OP_WRITE) != 0;
10
11        // return task to complete the job
12
13        Runnable task = fillable
14
15            ? (flushable
16
17            ? _runCompleteWriteFillable
18
19            : _runFillable)
20
21            : (flushable
22
23            ? _runCompleteWrite
24
25            : null);
26
27            
28
29        return task;
30
31    }
32

很简单，依据 selector 监听到连接上是读就绪（channel通道中有数据可读），还算写就绪（channel通道中有数据可写）。

Connection (HttpConnection)

这里如果是读，则进来 AbstractConnection$ReadCallback ，即是 HttpConnection 的 onFillable() 方法

调用 EndPoint 的接口去读数据，读完后让 HTTP 解析器HttpParser去解析字节流，HTTP 解析器会将解析后的数据，包括请求行、请求头相关信息存到 Request 对象里, 然后丢给我们的第一个Handler。

//HttpConnection.java

public void onFillable(){

1// HTTP 解析器去解析字节流
2
3boolean handle = parseRequestBuffer();
4
5...
6
7// 将解析后的数据，包括请求行、请求头相关信息存到 Request 对象, 然后丢给我们的 第一个Handler
8
9if (handle){
10
11boolean suspended = !_channel.handle();
12
13...

}

//HttpChannel.java

public boolean handle(){

...

dispatch(DispatcherType.REQUEST, () -> {

1        for (HttpConfiguration.Customizer customizer : _configuration.getCustomizers()) {
2
3            customizer.customize(getConnector(), _configuration, _request);
4
5            if (_request.isHandled())
6
7                return;
8
9        }
10
11        getServer().handle(HttpChannel.this);
12
13    });

....

}

比如HttpChannelOverHttp （HttpChannel的具体实现类之一）

处理完后， AbstractConnection$SendCallback调用 EndPoint的写方法