Vert.x 源码阅读 (4) - Context

Vert.x 源码阅读 (1) - Future 和 Promise

Vert.x 源码阅读 (2) - Stream

Vert.x 源码阅读 (3) - EventBus

Vert.x 源码阅读 (4) - Context

这是 Vert.x 项目源码阅读笔记的第四篇，主要记录一下 Vertx 中的核心 Context。Context 贯穿了整个 Vertx 的代码，它主要用来表示一个任务执行时的上下文环境。

一个服务实例往往需要同时处理大量的请求，而这些请求相互独立，拥有不同的上下文。在使用 Servlet 的年代，由于每个线程同时只处理一个请求，我们使用一个简单的 ThreadLocal 变量就可以满足需求。然而，在使用 EventLoop 的时候，由于一个线程会同时处理多个请求，我们需要显式地管理和切换上下文。Context 是 Vertx 中上下文的抽象。

简介

在详细介绍 Context 之前，我们先介绍几个名词：

Handler 是一个可执行的对象，类似 Runnable。
线程，就不多说了。
Execution 是指 Handler 的一次调用。Handler可以被多次调用。

Context 是

The execution context of a Handler execution.

简单来说，就是 Handler 的一次调用从开始到结束时，它使用的上下文信息。例如，一个 REST 请求执行过程中的 HEADER 信息。

在 Vertx 中，Context 和线程的关系简单来说，可以总结成一下几点

一个线程在不同的时间，会执行不同的 Handler，因此和它相关联的 context (通过 VertxThread::context) 获取会发生变化。即**一个线程会对应到多个 context**。
一个 Context 往往只对应到一个线程，但是并不强制。

Context 的继承关系如下图：

Context

根据上图，我们知道在 Vertx 主要有 EventLoopContext 和 WorkerContext 这两类 Context。这两类 Context，前一类对应到 EventLoop 线程，后一类对应到 Worker 线程。它们两个的区别主要在于，在执行/调度一个任务时（调用一个 Handler 时），到底使用哪一个线程。

`Context`

Context 是一个接口，它主要包含了一下几类方法：

获取当前执行线程的一些基本信息，例如 isOnWorkerThread。
获取自身属性，例如 isEventLoopContext。
读取或更新上下文信息，例如 get，put。
执行任务，例如 runOnContext ，executeBlocking。

前三类方法都相对比较好理解，我们主要看看 Context 提供的执行和调度任务的方法。

执行和调度任务

Context 主要提供了四种类型的调度方式：

方式名称	执行线程	上下文
`execute`	自己对应的线程	自己
`schedule`	自己对应的线程	线程执行任务时关联的 `context`
`emit`	调用 `emit` 方法的线程	自己
`dispatch`	自己对应的线程	自己

上表描述了各个类型的执行方法使用的线程和上下文，注意 dispatch 和 execute 的语义是一样的，它们的区别在于，

execute 将任务加入自己对应的线程的执行队列
dispatch 会判断当前线程是不是自己对应的线程，如果是的话直接执行。否则和 execute 一样。

上面的描述可能比较抽象，下面用伪代码来简单实现这几种方式：

// 通过保存和恢复现场，来实现在执行任务过程中，用自身作为上下文
public void emit(Runnable task) {
  // 获取当前线程使用的 context，并将当前线程关联的 context 设置为自己
  ContextInternal prev = emitBegin();
  try {
    handler.run();
  } catch (Throwable t) {
    reportException(t);
  } finally {
    // 恢复当前线程使用的 context
    emitEnd(prev);
  }
}

// 在自身关联的线程上，执行任务。这个任务被 emit 包裹，保证执行是会使用自身做上下文。
public void execute(Runnable task) {
  getAssociatedThread().execute(() -> emit(task));
}

// 在自身关联的线程上，执行任务。但是任务执行时的上下文没有单独指定。
public void schedule(T argument, Handler<T> task) {
  Thread thread = getAssociatedThread();
  if (thread.isCurrentThread()) {
    task.handle(argument);
  } else {
    thread.execute(() -> task.handle(argument));
  }
}

public void dispatch(T argument, Handler<T> task) {
  schedule(v -> emit(argument, task));
}

执行阻塞代码

在使用 EventLoop 时，我们不应该阻塞事件循环。这是因为事件循环的线程数很少，一旦阻塞了，新的请求都会被阻塞。当我们需要执行阻塞代码是，我们往往会使用另外的线程来执行。Context 提供了 executeBlocking 系列方法，来满足这个需求。它的实现伪代码如下：

@Override
public <T> Future<T> executeBlocking(Handler<Promise<T>> blockingCode, TaskQueue queue) {
  // Worker Pool 就是我们所说的非事件循环线程
  return executeBlocking(this, blockingCode, workerPool, queue);
}

static <T> Future<T> executeBlocking(ContextInternal context, 
                                     Handler<Promise<T>> blockingCode,
                                     WorkerPool workerPool, TaskQueue queue) {
  // 创建一对 Promise 和 Future
  // Promise 会传个 blockingCode. BlockingCode 在完成后应该调用 promise 的 complete 方法
  // 调用者使用 Future 来监听 blockingCode 是否完成
  Promise<T> promise = context.promise();
  Future<T> fut = promise.future();
  
  // 构造一个任务
  Runnable command = () -> {
    // 使用 emit 方法，保证会使用 context 作为上下文
    context.emit(promise, f -> {
      blockingCode.handle(promise);
    });
  };
  
  // 使用 workerPool 和 queue 执行任务
  queue.execute(command, exec);
  return fut;
}

看到上面的代码，我们可以发现，指定的阻塞任务

执行在 Worker 线程上
使用了当前的 (EventLoop) 的 Context

这样实现看似有点奇怪，实际上满足了等效于在事件循环上”执行“阻塞代码的要求。这也是为什么 Context 会提供单独的 isEventLoopContext/isWorkerContext 和 isOnEventLoopThread/isOnWorkerThread方法。

阻塞时长监控

虽然我们允许使用 Context 执行阻塞代码，阻塞的时间仍然不能太久。Vertx 通过 BlockedThreadChecker 来监控线程的阻塞时间。

BlockedThreadChecker 的实现比较简单。它通过一个后台线程，定期的检查每个线程当前执行的任务的开始时间离现在有多久。如果超过了最大时间，就打印一条日志。

瞎扯

目录

Vert.x 源码阅读 (4) - Context

简介

`Context`

执行和调度任务

执行阻塞代码

阻塞时长监控