浅析 Jetty 中的线程优化思路

Connector 也就是 Jetty 的连接器组件，相比较 Tomcat 的连接器，Jetty 的连接器在设计上有自己的特点。

Jetty 的 Connector 支持 NIO 通信模型，NIO 模型中的主角是 Selector，Jetty 在 Java 原生 Selector 的基础上封装了自己的 Selector：ManagedSelector。

这个过程有两个线程在干活：一个是 I/O 事件检测线程、一个是 I/O 事件处理线程。

这两个线程是"生产者"和"消费者"的关系。

Jetty 的 Connector 将 I/O 事件的生产和消费放到同一个线程处理。

如果执行过程中线程不阻塞，操作系统会用同一个 CPU 核来执行这两个任务，这样既能充分利用 CPU 缓存，又可以减少线程上下文切换的开销。

ManagedSelector 本质上是一个 Selector，负责 I/O 事件的检测和分发。

为了方便使用，Jetty 在 Java 原生 Selector 的基础上做了一些扩展，它的成员变量如下：

public class ManagedSelector extends ContainerLifeCycle implements Dumpable
{
    // 原子变量，表明当前的ManagedSelector是否已经启动
    private final AtomicBoolean _started = new AtomicBoolean(false);
     
    // 表明是否阻塞在select调用上
    private boolean _selecting = false;
     
    // 管理器的引用，SelectorManager管理若干ManagedSelector的生命周期
    private final SelectorManager _selectorManager;
     
    // ManagedSelector的id
    private final int _id;
     
    // 关键的执行策略，生产者和消费者是否在同一个线程处理由它决定
    private final ExecutionStrategy _strategy;
     
    // Java原生的Selector
    private Selector _selector;
     
    // "Selector更新任务"队列
    private Deque<SelectorUpdate> _updates = new ArrayDeque<>();
    private Deque<SelectorUpdate> _updateable = new ArrayDeque<>();
     
    ...
}

为什么需要一个"Selector更新任务"队列呢？

对于 Selector 的用户来说，我们对 Selector 的操作无非是将 Channel 注册到 Selector 或者告诉 Selector 我对什么 I/O 事件感兴趣。

这些操作其实就是对 Selector 状态的更新，Jetty 把这些操作抽象成 SelectorUpdate 接口。

/**
 * A selector update to be done when the selector has been woken.
 */
public interface SelectorUpdate
{
    void update(Selector selector);
}

这意味着不能直接操作 ManagedSelector 中的 Selector，而是需要向 ManagedSelector 提交一个任务类。

这个类需要实现 SelectorUpdate 接口的 update 方法，在 update 方法中定义要对 

ManagedSelector 做的操作。

比如 Connector 中的 Endpoint 组件对读就绪事件感兴趣。

它就向 ManagedSelector 提交了一个内部任务类

ManagedSelector.SelectorUpdate：

_selector.submit(_updateKeyAction);

private final ManagedSelector.SelectorUpdate _updateKeyAction = new ManagedSelector.SelectorUpdate()
{
    @Override
    public void update(Selector selector)
{
        // 这里的updateKey其实就是调用了SelectionKey.interestOps(OP_READ);
        updateKey();
    }
};

上面有了 update 方法，那谁来执行这些 update 呢，答案是 ManagedSelector 自己。

它在一个死循环里拉取这些 SelectorUpdate 任务逐个执行。

I/O 事件到达时，ManagedSelector 通过一个任务类接口（Selectable 接口）来确定由哪个函数处理这个事件。

public interface Selectable
{
    // 当某一个Channel的I/O事件就绪后，ManagedSelector会调用的回调函数
    Runnable onSelected();
 
    // 当所有事件处理完了之后ManagedSelector会调的回调函数
    void updateKey();
}

Selectable 接口的 onSelected() 方法返回一个 Runnable，这个 Runnable 就是 I/O 事件就绪时相应的处理逻辑。

ManagedSelector 在检测到某个 Channel 上的 I/O 事件就绪时，ManagedSelector 调用这个 Channel 所绑定的类的 onSelected 方法来拿到一个 Runnable。

然后把 Runnable 扔给线程池去执行。

1. 如何注册 Channel 以及 I/O 事件

2. 提供什么样的处理类来处理 I/O 事件

那么 ManagedSelector 如何统一管理和维护用户注册的 Channel 集合呢，答案是

ExecutionStrategy 接口。

这个接口将具体任务的生产委托给内部接口 Producer，而在自己的 produce 方法里实现具体执行逻辑。

这个 Runnable 的任务可以由当前线程执行，也可以放到新线程中执行。

public interface ExecutionStrategy
{
    // 只在HTTP2中用到的一个方法，暂时忽略
    public void dispatch();
 
    // 实现具体执行策略，任务生产出来后可能由当前线程执行，也可能由新线程来执行
    public void produce();
     
    // 任务的生产委托给Producer内部接口
    public interface Producer
    {
        // 生产一个Runnable（任务）
        Runnable produce();
    }
}

private class SelectorProducer implements ExecutionStrategy.Producer
{
    private Set<SelectionKey> _keys = Collections.emptySet();
    private Iterator<SelectionKey> _cursor = Collections.emptyIterator();
 
    @Override
    public Runnable produce()
{
        while (true)
        {
            // 如果Channel集合中有I/O事件就绪，调用前面提到的Selectable接口获取Runnable,直接返回给ExecutionStrategy去处理
            Runnable task = processSelected();
            if (task != null)
                return task;
             
           // 如果没有I/O事件就绪，就干点杂活，看看有没有客户提交了更新Selector的任务，就是上面提到的SelectorUpdate任务类。
            processUpdates();
            updateKeys();
 
           // 继续执行select方法，侦测I/O就绪事件
            if (!select())
                return null;
        }
    }
 }

SelectorProducer 是 ManagedSelector 的内部类。

SelectorProducer 实现了 ExecutionStrategy 中的 Producer 接口中的 produce 方法，需要向 ExecutionStrategy 返回一个 Runnable。

在 produce 方法中 SelectorProducer 主要干了三件事：

任务生产者自己依次生产和执行任务，对应到 NIO 通信模型就是用一个线程来侦测和处理一个 ManagedSelector 上的所有的 I/O 事件。

后面的 I/O 事件要等待前面的 I/O 事件处理完，效率明显不高。

任务生产者开启新线程来执行任务，这是典型的 I/O 事件侦测和处理用不同的线程来处理。

缺点是不能利用 CPU 缓存，并且线程切换成本高。

任务生产者自己运行任务，这种方式可能会新建一个新的线程来继续生产和执行任务。

它的优点是能利用 CPU 缓存，但是潜在的问题是如果处理 I/O 事件的业务代码执行时间过长，会导致线程大量阻塞和线程饥饿。

这是 Jetty 对 ExecuteProduceConsume 策略的改良，在线程池线程充足的情况下等同于 ExecuteProduceConsume；

当系统比较忙线程不够时，切换成

ProduceExecuteConsume 策略。

参考文档：