Vert.x学习笔记-什么是Verticle

什么是Verticle

Verticle是Vert.x应用中的基本编程单元，类似于Java中的Servlet、Pojo Bean或Akka中的Actor。它可以使用不同的编程语言实现，并且这些由不同编程语言实现的Verticle可以封装到一个模块中，进而部署到一个Vert.x应用中。Verticle可分为两种类型：标准Verticle和Worker Verticle。

• 标准Verticle运行在Vert.x实例的事件循环线程中，当有事件发生时，在事件循环线程中回调Verticle实例中的event handler。
• Worker Verticle在background workers线程池中执行，该线程池的大小缺省为40。

在事件循环线程上运行代码的基本原则是不能阻塞，并且代码的执行要足够快，但是在很多场合中确实难以避免会有阻塞的代码出现，vert.x提供了两种处理此类情况的方法： Worker Verticle 和 executeBlocking方法

Worker Verticle

Worker Verticle是一种特殊的Verticle，它不是在事件循环线程上执行的，而是在专门的工作线程上执行的。工作线程来自一个特殊的工作线程池，开发人员也可以自定义工作线程池，将Worker Verticle部署给工作线程池。

Worker Verticle处理事件的方式与事件循环一致，只不过它可用人异常的时间来处理事件，Worker Verticle有以下两个特点：

• Worker Verticle不会与某一个工作线程绑定，每一次运行可能是在不同的工作线程上的
• 在任意的时刻，Worker Verticle只能被单个工作线程访问

Working Verticle和事件循环一样，都是单线程的，不同的地方是事件循环线程固定，Worker Verticle依附的线程不固定，每次可能会从线程池中找新的线程处理

Workder Verticle示例

public class WorkerVerticle extends AbstractVerticle{

  private final Logger logger = LoggerFactory.getLogger(WorkerVerticle.class);

  @Override
  public void start() {
    vertx.setPeriodic(10_000, id -> {
      try {
        logger.info("Zzz...");
        Thread.sleep(8000);
        logger.info("Up!");
      } catch (InterruptedException e) {
        logger.error("Woops", e);
      }
    });
  }

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    DeploymentOptions opts = new DeploymentOptions()
      .setInstances(2) //创建两个实例
      .setWorker(true); //通过该标志来确定实例化的是否为Worker Verticle
    vertx.deployVerticle("xxx.WorkerVerticle", opts); //由于单次要创建两个实例，所以需要用全路径，如果只是创建单个Verticle就可以用new或者全路径
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

从代码的执行结果也可以明显的看出来，每一次执行Worker Verticle的并不是固定的线程

executeBlocking方法

虽说Worker Verticle处理阻塞任务是一个非常优秀的方案，但是将阻塞代码全部抽取到Worker Verticle也不一定是完全合理的，这样可能会造成有非常多的执行很小功能的Worker Verticle，并且造成各个类无法形成一个独立的功能单元
运行阻塞式代码的另一种方式是通过Vertx类中的executeBlocking方法。该方法将一些阻塞的代码的执行转移到工作线程上，然后会以新事件的形式将结果发送回事件循环，执行流程如下图所示：

代码示例

public class Offload extends AbstractVerticle {

  private final Logger logger = LoggerFactory.getLogger(Offload.class);

  @Override
  public void start() {
    vertx.setPeriodic(5000, id -> {
      logger.info("Tick");
      vertx.executeBlocking(this::blockingCode, this::resultHandler);
    });
  }

  private void blockingCode(Promise<String> promise) { // promise对象用于返回结果的传递
    logger.info("Blocking code running");
    try {
      Thread.sleep(4000);
      logger.info("Done!");
      promise.complete("Ok!"); //promise完成并返回OK
    } catch (InterruptedException e) {
      promise.fail(e); //promise失败并返回异常
    }
  }

  private void resultHandler(AsyncResult<String> ar) { //在事件循环线程上处理阻塞代码的结果
    if (ar.succeeded()) {
      logger.info("Blocking code result: {}", ar.result());
    } else {
      logger.error("Woops", ar.cause());
    }
  }

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle(new Offload());
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

执行结果：

从上图的执行 结果就可以看出来，事件循环线程是固定不变的，但是执行executeBlocking方法的线程每次执行都可能是会变化的，并且是worker线程来进行执行的

默认情况下，当Vert.x有连续的多个executeBlocking方法时，会按照其调用顺序获得执行结果，executeBlocking方法还有一个带布尔参数类型的重载（ Future<@Nullable T> executeBlocking(Handler blockingCodeHandler, boolean ordered);），当该参数被设置为false时，任务执行完成后立即会将结果返回给事件循环，而不再遵循executeBlocking的调用顺序

Verticle 运行环境有Context

根据上图，我们来详细阐述一下Verticle相关的一些内容
Verticle对象本质是两个对象的组合，分别是：Vert.x和Context

• Vertx对象： 该Verticle对象所属的Vert.x实例
• Context对象： 记录Verticle上下文信息的实例，Verticle借助Context对象将事件分派到各个处理程序中

Vert.x实例提供了许多核心API来声明事件处理程序

Vert.x实例由多个Verticle共享，通常每个JVM程序只需要一个Vert.x实例

Verticle Context 可访问此Verticle的事件处理程序的执行线程，事件的来源有很多，比如数据库驱动程序、定时器程序、Http服务器程序，实际环境中，事件更多的是由其它线程触发的，例如Netty接收连接的线程
用户自定义的事件处理程序借助Verticle Context运行，Verticle Context使得我们可以在Verticle分配到的事件循环上执行事件处理程序，从而遵循Vert.x的线程模型。

Worker Verticle 与 Context

Worker Verticle事件处理程序由工作线程池中的某个工作线程执行，初次之外，Worker Verticle与事件循环的情况基本一致

Context

Verticle Context是Vert.x框架中的一种机制，用于管理和执行Verticle。当Vert.x提供一个事件的处理程序或调用Verticle的开始或停止的方法时，执行与Context相关联。在Verticle中，Context通常是事件循环 context绑定特定的事件循环线程。因此，对于这方面的执行总是发生在该完全相同的事件循环线程。在worker verticles和运行内嵌阻塞代码worker context的情况下将使用一个线程从worker线程池的执行关联。

Verticle的Context对象可以通过Vert.x的getOrCreateContext（）方法获取，虽然Context对象几乎都是与Verticle相关联的，单实际上也可以在Verticle之外创建Context对象

• 从Vert.x上下文环境中调用 getOrCreateContext（）方法时，会返回当前的Context对象，如Verticle Context
• 从非Vert.x上下文环境中调用getOrCreateContext（）方法时，会创建一个新的Context对象
  下面的代码就是在非Vert.x上下文环境（JVM进程的主线程）中创建了Context

public class ThreadsAndContexts {

  private static final Logger logger = LoggerFactory.getLogger(ThreadsAndContexts.class);

  public static void main(String[] args) {
    createAndRun();
    dataAndExceptions();
  }

  private static void createAndRun() {
    Vertx vertx = Vertx.vertx();

    vertx.getOrCreateContext()
      .runOnContext(v -> logger.info("ABC")); //Lambda表达式在Vert.x上下文线程上执行

    vertx.getOrCreateContext()
      .runOnContext(v -> logger.info("123"));
  }

  private static void dataAndExceptions() {
    Vertx vertx = Vertx.vertx();
    Context ctx = vertx.getOrCreateContext();
    ctx.put("foo", "bar"); //上下文中可以保存任意的键值对

    ctx.exceptionHandler(t -> { //还可以声明一个异常处理程序
      if ("Tada".equals(t.getMessage())) {
        logger.info("Got a _Tada_ exception");
      } else {
        logger.error("Woops", t);
      }
    });

    ctx.runOnContext(v -> {
      throw new RuntimeException("Tada"); 
    });

    ctx.runOnContext(v -> {
      logger.info("foo = {}", (String) ctx.get("foo"));
    });
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

当事件处理程序的实现跨越了多个类时，不妨用Context来携带数据，这样可能比较有用，否则，不如直接使用某个类的变量

如果某个事件处理程序可能抛出异常，则异常处理程序会变得非常重要。

桥接Vert.x线程和非Vert.x线程

在编写Vert.x代码的时候，通常不需要同Vert.x的Context打交道，但有一种情况例外：我们不得不使用某些第三方代码，这些代码内置了自己的线程模型，而我们又想让他们与Vert.x协同工作，下面是一个非常常用的解决方法：

public class MixedThreading extends AbstractVerticle {

  private final Logger logger = LoggerFactory.getLogger(MixedThreading.class);

  @Override
  public void start() {
    Context context = vertx.getOrCreateContext(); //提前获取Verticle的Context
    new Thread(() -> {
      try {
        run(context);
      } catch (InterruptedException e) {
        logger.error("Woops", e);
      }
    }).start();
  }

  private void run(Context context) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(1);
    logger.info("I am in a non-Vert.x thread");
    context.runOnContext(v -> { //runOnContext方法确保这些代码在事件循环线程上执行的
      logger.info("I am on the event-loop");
      vertx.setTimer(1000, id -> {
        logger.info("This is the final countdown");
        latch.countDown();
      });
    });
    logger.info("Waiting on the countdown latch...");
    latch.await();
    logger.info("Bye!");
  }

  public static void main(String[] args) {
    Vertx vertx = Vertx.vertx();
    vertx.deployVerticle(new MixedThreading());
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

上面的代码创建了一个非Vert.x线程，先从Verticle上获取Context对象，再将它传递给非Vert.x线程，以便在非Vert.x线程环境中调用时间循环线程来执行一些代码

总结

• Verticle是Vert.x应用程序中处理异步事件的核心组件
• 事件循环负责处理异步的IO事件，它不应该被阻塞，也不应该用来处理耗时很长的任务
• Worker Verticle可用于处理阻塞的IO事件，或者耗时的任务
• 借助事件循环上下文，开发人员可以在代码中混合使用Vert.x线程和非Vert.x线程

拓展

下面是与Vert.x相关的一些其它博文链接，希望可以帮助大家进一步的了解Vert.x的相关知识

Vert.x学习笔记-异步编程和响应式系统

Vert.x学习笔记-什么是Vert.x

Vert.x学习笔记-Vert.x的基本处理单元Verticle