Vert.x 线程模型揭秘

Vert.x是一个在JVM开发reactive应用的框架，可用于开发异步、可伸缩、高并发的Web应用(虽然不限于web应用)。其目的在于为JVM提供一个Node.js的替代方案。开发者可以通过它使用JavaScript、Ruby、Groovy、Java，甚至是混合语言来编写应用。使用Vertx.x框架，可以用JavaScript、CoffeeScript、Ruby、Python、Groovy或Java开发应用程序的组件，最终应用程序可以是混合语言构建的。

本文试图揭示Vert.x的线程模型的应用，通过源代码分析Vert.x如何使用线程池处理请求的，以及比较Netty和Vert.x使用线程池的异同。

也许你觉得奇怪，默认启动一个Vert.x Verticle实例，它只用一个线程处理事件，在多核的情况下你需要创建多个Verticle实例以充分利用多个CPU Core的性能。

Vert.x 实例

首先先啰嗦地介绍一些Vert.x概念，熟悉Vert.x开发的朋友可以跳过这一节。

在Vert.x里，如果你不使用Vertx对象，你几乎是寸步难行。

Vertx对象扮演着Vert.x控制中心的角色，同时它也提供了大量的功能，例如：

• 编写TCP客户端和服务器
• 编写HTTP客户端和服务器，包括websocket
• Event bus
• 共享数据
• 定时器
• 发布和卸载Verticle
• UDP
• DNS client
• 文件系统访问
• 高可用
• 集群

如果你将Vert.x嵌入到你的应用程序中，你可以向下面这样获得一个Vertx对象的引用

Vertx vertx = Vertx.vertx(); 

当你实例化Vertx对象时，如果你感觉默认的参数不符合你的需求，你可以指定实例化时的参数：

Vertx vertx = Vertx.vertx(newVertxOptions().setWorkerPoolSize(40)); 

VertxOptions 对象拥有很多关于Vertx实例设置，例如配置集群，高可用设置，线程池大小以及等等其他参数。下面就介绍一下它的线程池。

线程池

eventLoopThreadFactory = newVertxThreadFactory("vert.x-eventloop-thread-", checker,false);  eventLoopGroup = newNioEventLoopGroup(options.getEventLoopPoolSize(), eventLoopThreadFactory);  eventLoopGroup.setIoRatio(NETTY_IO_RATIO); 

它的 EventLoop 和一个Context对应：

protectedContextImpl(……) {  ……  EventLoopGroup group = vertx.getEventLoopGroup(); if(group !=null) { this.eventLoop = group.next();  } else{ this.eventLoop =null;  }  …… } 

它用来执行标准的Verticle。

2、WorkerPool用来执行worker Verticle。

workerPool = Executors.newFixedThreadPool(options.getWorkerPoolSize(), newVertxThreadFactory("vert.x-worker-thread-", checker,true)); 

3、Internal Blocking Pool内部使用的线程池，可以用来将阻塞代码异步化。

internalBlockingPool = Executors.newFixedThreadPool(options.getInternalBlockingPoolSize(), newVertxThreadFactory("vert.x-internal-blocking-", checker,true)); 

不要在event loop中执行阻塞操作， 比如访问数据库或者网络资源，这绝对会影响你的应用的性能。对于这些阻塞操作，你可以将它们异步化：

vertx.executeBlocking(future -> { // 下面这行代码可能花费很长时间  String result = someAPI.blockingMethod("hello");  future.complete(result); }, res -> {  System.out.println("The result is: "+ res.result()); }); 

默认情况下executeBlocking会在同一个context中执行(同一个verticle实例)，它们会串行化执行。如果不关心这个执行的顺序，可以将ordered参数设为false，它们会在worker pool线程池中并行的执行。

另外一种执行阻塞代码的方式就是使用worker verticle，worker verticle总是在worker pool线程池中执行。

Verticle

Verticle有点类似Actor模型，也可以实现并发的，可扩展的，易于发布的模型。

一个vert.x应用可以包含多个verticle实例，实例之间可以通过event bus通讯。

三种类型

http://vertx.io/docs/vertx-core/java/#_verticle_types

实现一个Verticle很简单：

publicclassMyVerticleextendsAbstractVerticle{  // 当发布verticle时调用 publicvoidstart() {  }  // 可以不实现。当 verticle 卸载时调用 publicvoidstop() {  }  } 

发布方式

<transformers> <transformerimplementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer"> <manifestEntries> <Main-Class>io.vertx.core.Starter</Main-Class> <Main-Verticle>com.colobu.vertx.Main</Main-Verticle> </manifestEntries> </transformer> <transformerimplementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"> <resource>META-INF/services/io.vertx.core.spi.VerticleFactory</resource> </transformer> </transformers> 

然后运行 java -jar xxx-fat.jar ,你还可以传递一些参数。

publicclassMainextendsAbstractVerticle{  publicstaticvoidmain(String[] args) {  VertxOptions vo = newVertxOptions();  vo.setEventLoopPoolSize(16);  Vertx vertx = Vertx.vertx(vo);   DeploymentOptions options = newDeploymentOptions();  options.setInstances(100);   vertx.deployVerticle(Main.class.getName(), options, e -> {  System.out.println(e.succeeded());  System.out.println(e.failed());  System.out.println(e.cause());  System.out.println(e.result());  });  } @Override publicvoidstart() {  Handler<HttpServerRequest> handler = e -> {  HttpServerResponse response = e.response();  response.putHeader("content-type","application/json").end("Hello world");  };  vertx.createHttpServer().requestHandler(handler).listen(8080);  } } 

Verticle发布和Vert.x线程模型

以上比较啰嗦，主要介绍了一些Vert.x的一些概念。下面是我想重点介绍的内容。

本节以实现一个简单的http server为例(编程方式发布Verticle)，分析 vert.x 的线程和Verticle的关系。只分析标准的Verticle。代码如上。

Verticle发布过程

首先先创建一个Vertx实例，可以你可以通过 VertxOptions 设置线程池的大小。上面的例子中设置Event Loop线程池的大小为16:

vo.setEventLoopPoolSize(16); 

因此即使你创建几百个Verticle，也只会有16个Event Loop处理它们，你可以通过 jstack 查看这些线程。你会看到多个名为 vert.x-eventloop-thread-<num> 的线程，一个 vertx-blocked-thread-checker 线程，一个 vert.x-acceptor-thread-0 。

底层调用 DeploymentManager 的 doDeployVerticle 来实现，它会根据实例数创建相应多的 Verticle ,然后调用 doDeploy 发布这些 Verticle :

Verticle[] verticles = createVerticles(verticleFactory, identifier, options.getInstances(),cl); 

我将 doDeploy 方法简化，让我们看一下关键代码：

privatevoiddoDeploy(String identifier, String deploymentID, DeploymentOptions options,  ContextImpl parentContext,  ContextImpl callingContext,  Handler<AsyncResult<String>> completionHandler,  ClassLoader tccl, Verticle... verticles) { //准备工作  ……  for(Verticle verticle: verticles) { //创建上下文  ContextImpl context = options.isWorker() ? vertx.createWorkerContext(options.isMultiThreaded(), deploymentID, conf, tccl) :  vertx.createEventLoopContext(deploymentID, conf, tccl);   deployment.addVerticle(newVerticleHolder(verticle, context));  context.runOnContext(v -> { try{  verticle.init(vertx, context);  Future<Void> startFuture = Future.future();  verticle.start(startFuture);  startFuture.setHandler(……);  } catch(Throwable t) {}  });  }  } 

可以看到#11 行创建了一个上下文ContextImpl, 因为本例中我们不用worker模式，所以这个上下文是通过 vertx.createEventLoopContext(deploymentID, conf, tccl) 创建的。每个verticle都会创建一个新的上下文，因此verticle和上下文是意义对应的。

#17 行初始化verticle，#19 行启动这个verticle。还记得我们的例子中实现的 start 方法吗，它会在这里被调用。

这样，多个verticle实例被发布了。

线程模型

首先插播一下Netty的线程模型，不感兴趣的可以略过。

Netty的线程模型

EventLoopGroup parentGroup = newNioEventLoopGroup(1); EventLoopGroup childGroup = newNioEventLoopGroup(50); try{  ServerBootstrap b = newServerBootstrap();  b.group(parentGroup, childGroup)  .channel(NioServerSocketChannel.class)  .childHandler(newChannelInitializer<SocketChannel>(){……});   Channel ch = b.bind("0.0.0.0",8080).sync().channel();  ch.closeFuture().sync();  } finally{  parentGroup.shutdownGracefully();  childGroup.shutdownGracefully();  } 

当有多个并发连接的时候，每个连接/Channel被分配到一个 EventLoop 上。 EventLoop 选择是均匀地 (如果线程数是2的n次方，可以用比较快的选择方法PowerOfTwoEventExecutorChooser):

privatefinalclassPowerOfTwoEventExecutorChooserimplementsEventExecutorChooser{ @Override publicEventExecutornext() { returnchildren[childIndex.getAndIncrement() & children.length -1];  } }  privatefinalclassGenericEventExecutorChooserimplementsEventExecutorChooser{ @Override publicEventExecutornext() { returnchildren[Math.abs(childIndex.getAndIncrement() % children.length)];  } } 

publicclassHelloServerHandlerextendsChannelInboundHandlerAdapter{  ……  @Override publicvoidchannelRead(ChannelHandlerContext ctx, Object msg) {  String name = Thread.currentThread().getName();  System.out.println(name); if(name.endsWith("-5")) { try{  Thread.sleep(1000);  } catch(InterruptedException e) {  e.printStackTrace();  }  }  ……  ｝ 

输出结果可以看到 nioEventLoopGroup-3-5 处理了同样多的请求，而且都堆积在后面了。

…… nioEventLoopGroup-3-19 nioEventLoopGroup-3-18 nioEventLoopGroup-3-19 nioEventLoopGroup-3-18 nioEventLoopGroup-3-20 nioEventLoopGroup-3-20 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 nioEventLoopGroup-3-5 

因此，我们可以了解到，当启动一个NIO方式的Netty实例的时候，它会使用一个线程池来处理http请求。

Netty 4.0的线程模型被很好的重定义，一个 ChannelHandler 实例的方法不会被并发的调用，除非它被 @Sharable 标记，因此你不应该增加一个ChannelHandler 实例多次。当你增加一个handler到ChannelPipeline中时，你可以指定一个特定的 EventExecutorGroup 来执行这个handler。如果没有指定，则使用Channel注册的 EventLoop 来执行。如果两个Handler被指定不同的 EventExecutorGroup ，则它们会并发执行，因此如果它们会访问共享数据的化，你需要关注并发控制的问题。更多内容可以查看 Netty的文档 。

Vert.x的线程模型

Vert.x如何在线程中处理事件的呢，还是以我们的例子分析。

回顾一下我们实现的Verticle的start方法。

@Override publicvoidstart() {  Handler<HttpServerRequest> handler = e -> {  HttpServerResponse response = e.response();  response.putHeader("content-type","application/json").end("Hello world");  };  vertx.createHttpServer().requestHandler(handler).listen(8080);  } 

在这个start方法中，我们创建了一个http server，让它监听 8080端口， http request的处理交给handler执行。 那么监听线程是哪一个？handler又是在哪个线程池中执行的呢？调用多个Verticle实例的方法为什么没有出现"地址/端口被占用"的异常呢？

首先vertx.createHttpServer()会创建一个HttpServerImpl对象，可以通过HttpServerOptions配置更多的参数，每个Verticle实例都会创建一个HttpServerImpl对象。requestHandler(handler)方法设置处理器，你还可以使用Vert.x-Web设置路由的功能。

publicsynchronizedHttpServerlisten(intport, String host, Handler<AsyncResult<HttpServer>> listenHandler) {  listenContext = vertx.getOrCreateContext();  listening = true; synchronized(vertx.sharedHttpServers()) {  id = newServerID(port, host);  HttpServerImpl shared = vertx.sharedHttpServers().get(id); if(shared ==null) {  serverChannelGroup = newDefaultChannelGroup("vertx-acceptor-channels", GlobalEventExecutor.INSTANCE);  ServerBootstrap bootstrap = newServerBootstrap();  bootstrap.group(vertx.getAcceptorEventLoopGroup(), availableWorkers);  bootstrap.channelFactory(newVertxNioServerChannelFactory());   bootstrap.childHandler(newChannelInitializer<Channel>() { @Override protectedvoidinitChannel(Channel ch)throwsException {  ……  pipeline.addLast("handler",newServerHandler());  }  });   addHandlers(this, listenContext);   vertx.sharedHttpServers().put(id, this);  actualServer = this;  } else{ // Server already exists with that host/port - we will use that  actualServer = shared;  addHandlers(actualServer, listenContext);  metrics = vertx.metricsSPI().createMetrics(this,newSocketAddressImpl(port, host), options);  }   } returnthis;  } 

#6 行可以看到它会检查使用这个IP地址和端口的http server是否存在，如果存在的化直接跳到# 27行。因此回答上面的问题，多个Verticle实例不会引起冲突，因为它们会共享同一个http server。

这个http server通过netty ServerBootstrap创建。#10 行可以看到acceptor是一个单线程执行的,acceptorEventLoopGroup在VertxImpl中定义。

acceptorEventLoopGroup =new NioEventLoopGroup(1, acceptorEventLoopThreadFactory); 

#10 行还显示，netty的IO worker线程池由 availableWorkers 确定，它是一个VertxEventLoopGroup对象。VertxEventLoopGroup类扩展 AbstractEventExecutorGroup ,实现了 EventLoopGroup 接口:

…… @Override publicsynchronizedEventLoopnext() { if(workers.isEmpty()) { thrownewIllegalStateException();  } else{  EventLoop worker = workers.get(pos).worker;  pos++;  checkPos(); returnworker;  } }  publicsynchronizedvoidaddWorker(EventLoop worker) {  EventLoopHolder holder = findHolder(worker); if(holder ==null) {  workers.add(newEventLoopHolder(worker));  } else{  holder.count++;  } }  …… 

线程的数量由 worker 的数量决定，worker的类型是EventLoop,对应一个线程，有多少 worker 就会有多少线程。

通过 addWorker 可以增加线程的数量，worker不会重复。

回到刚才的 listen 方法， #21 行addHandlers方法会配置handler在哪一个event loop中执行：

privatevoidaddHandlers(HttpServerImpl server, ContextImpl context) { if(requestStream.handler() !=null) {  server.reqHandlerManager.addHandler(requestStream.handler(), context);  } if(wsStream.handler() !=null) {  server.wsHandlerManager.addHandler(wsStream.handler(), context);  } } 

server.reqHandlerManager.addHandler 方法如下：

publicsynchronizedvoidaddHandler(Handler<T> handler, ContextImpl context) {  EventLoop worker = context.nettyEventLoop();  availableWorkers.addWorker(worker);  Handlers<T> handlers = newHandlers<>();  Handlers<T> prev = handlerMap.putIfAbsent(worker, handlers); if(prev !=null) {  handlers = prev;  }  handlers.addHandler(newHandlerHolder<>(context, handler));  hasHandlers = true; } 

因此我们可以看出正常情况下Vert.x的每个 Verticle 实例只会用一个线程处理请求，在多核情况下一定要配置instance的数量。

如果配置的instance的数量大于eventLoopPoolSize数量，那么就会有一个Event Loop处理多个instance的情况。 线程配置的过多有时不会带来性能的提升，由于线程也有context swicthing，反而会带来性能的降低。