vertx源码分析（二）---------------vertx-web router如何部署到vertx中

朴灿烈づ我的快乐病毒、 2023-06-12 11:22 27阅读 0赞

在分析一中，我们已经知道了verticle是如何部署的  
而route就是实现了AbstractVerticle的一种

### route ###

package cn.ctyun.moho.router.verticle;
    
    import cn.ctyun.moho.router.config.AppConfig;
    import cn.ctyun.moho.router.handler.*;
    import cn.ctyun.moho.router.model.PluginType;
    import cn.ctyun.moho.router.plugin.IPluginHandler;
    import io.vertx.core.AbstractVerticle;
    import io.vertx.core.http.HttpServerOptions;
    import io.vertx.ext.web.Router;
    import io.vertx.ext.web.handler.BodyHandler;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Scope;
    import org.springframework.stereotype.Component;
    
    import java.util.Comparator;
    import java.util.List;
    
    /**
     * 负责路由转发的核心Vertical
     * @author Codi
     */
    @Slf4j
    @Component
    @Scope("prototype")
    public class RouterVerticle extends AbstractVerticle {
    
        @Autowired
        private RouteHandler routeHandler;
        @Autowired
        private BlockerHandler blockerHandler;
        @Autowired
        private RequestHandler requestHandler;
        @Autowired
        private ResponseHandler responseHandler;
        @Autowired
        private HttpClientHandler httpClientHandler;
        @Autowired
        private List<IPluginHandler> pluginHandlers;
        @Autowired
        private ReturnHandler returnHandler;
        @Autowired
        private GlobalErrorHandler errorHandler;
        @Autowired
        private AppConfig appConfig;
        @Autowired
        private StopHandler stopHandler;
        //从之前的分析可以知道，vertx在初始化完成后，会运行注册的AbstractVerticle
        //中的start方法
        @Override
        public void start() {
            Router router = Router.router(vertx);
            //处理请求body
            //TODO 对异常body过滤
            router.route().handler(BodyHandler.create());
            //优雅关闭handler
            router.route().handler(stopHandler);
            //整理HttpRequest
            router.route().handler(requestHandler);
            //处理路由拦截
            router.route().handler(blockerHandler);
            //处理路由匹配
            router.route().handler(routeHandler);
            //处理前置插件
            initPreHandlers(router);
            //处理转发 整理response
            router.route().handler(httpClientHandler);
            //处理后置插件
            initPostHandlers(router);
            //处理返回 这里调用last方法会改变route中order的值，让后边的handler放到最后
            router.route().last().handler(returnHandler);
            //集中异常处理
            router.route().failureHandler(errorHandler);
            //启动http服务 这个才是启动route的关键
            startHttpServer(router);
        }
    
        private void initPreHandlers(Router router) {
            pluginHandlers.stream()
                    .filter(handler -> handler.getPluginType() == PluginType.PRE)
                    .sorted(Comparator.comparing(IPluginHandler::priority))
                    .forEach(handler -> {
                        router.route().handler(handler);
                        log.info("Add pre plugin <" + handler.getClass().getSimpleName() + ">");
                    });
            log.info("PreHandlers init complete");
        }
    
        private void initPostHandlers(Router router) {
            pluginHandlers.stream()
                    .filter(handler -> handler.getPluginType() == PluginType.POST)
                    .sorted(Comparator.comparing(IPluginHandler::priority))
                    .forEach(handler -> {
                        router.route().handler(handler);
                        log.info("Add post plugin <" + handler.getClass().getSimpleName() + ">");
                    });
            log.info("PostHandlers init complete");
        }
    
        private void startHttpServer(Router router) {
            int port= appConfig.getPort();
            HttpServerOptions options = new HttpServerOptions();
            //增加配置初始长度，规避可能存在的uri过长被服务器拒绝的问题
            options.setMaxInitialLineLength(appConfig.getMaxInitialLineLength());
            vertx.createHttpServer(options).requestHandler(router).listen(port, http -> {
                if (http.succeeded()) {
                    log.info("Moho-Router HTTP server started on port " + port);
                } else {
                    log.error("Moho-Router HTTP server start error!", http.cause());
                }
            });
        }
    
    }

createHttpServer：

public HttpServer createHttpServer(HttpServerOptions serverOptions) {
        return new HttpServerImpl(this, serverOptions);
      }
      
      public HttpServerImpl(VertxInternal vertx, HttpServerOptions options) {
        this.options = new HttpServerOptions(options);
        this.vertx = vertx;
        //这个时整个route的全局context，原理和之前的vertx的context是一致的
        this.creatingContext = vertx.getContext();
        if (creatingContext != null) {
          if (creatingContext.isMultiThreadedWorkerContext()) {
            throw new IllegalStateException("Cannot use HttpServer in a multi-threaded worker verticle");
          }
          creatingContext.addCloseHook(this);
        }
        this.sslHelper = new SSLHelper(options, options.getKeyCertOptions(), options.getTrustOptions());
        this.subProtocols = options.getWebsocketSubProtocols();
        this.logEnabled = options.getLogActivity();
      }

requestHandler 将route的配置注入到vertx中：

@Override
      public synchronized HttpServer requestHandler(Handler<HttpServerRequest> handler) {
      //requeststream是创建httpserverImpl时初始化的
      //private final HttpStreamHandler<HttpServerRequest> requestStream = new HttpStreamHandler<>();
        requestStream.handler(handler);
        return this;
      }
      
       @Override
        public ReadStream handler(Handler<C> handler) {
          synchronized (HttpServerImpl.this) {
            if (listening) {
              throw new IllegalStateException("Please set handler before server is listening");
            }
            this.handler = handler;
            return this;
          }
        }

开始监听listen:

public synchronized HttpServer listen(SocketAddress address, Handler<AsyncResult<HttpServer>> listenHandler) {
        if (requestStream.handler() == null && wsStream.handler() == null) {
          throw new IllegalStateException("Set request or websocket handler first");
        }
        if (listening) {
          throw new IllegalStateException("Already listening");
        }
        //新建一个context给这个eventloop用
        listenContext = vertx.getOrCreateContext();
        listening = true;
        String host = address.host() != null ? address.host() : "localhost";
        int port = address.port();
        serverOrigin = (options.isSsl() ? "https" : "http") + "://" + host + ":" + port;
        List<HttpVersion> applicationProtocols = options.getAlpnVersions();
        if (listenContext.isWorkerContext()) {
          applicationProtocols =  applicationProtocols.stream().filter(v -> v != HttpVersion.HTTP_2).collect(Collectors.toList());
        }
        sslHelper.setApplicationProtocols(applicationProtocols);
        //如果host和端口一样，防止同时创建多个
        synchronized (vertx.sharedHttpServers()) {
          this.actualPort = port; // Will be updated on bind for a wildcard port
          id = new ServerID(port, host);
          HttpServerImpl shared = vertx.sharedHttpServers().get(id);
          if (shared == null || port == 0) {
            serverChannelGroup = new DefaultChannelGroup("vertx-acceptor-channels", GlobalEventExecutor.INSTANCE);
            //这一步就是netty的服务端创建的方式，详细分析可以自行谷歌netty服务端实现
            ServerBootstrap bootstrap = new ServerBootstrap();
            //第一个为bosseventloopgroup第二个为workeventloopgroup
            bootstrap.group(vertx.getAcceptorEventLoopGroup(), availableWorkers);
            applyConnectionOptions(address.path() != null, bootstrap);
            sslHelper.validate(vertx);
            //childhandler为第二个workeventloopgroup的配置，只有在有请求第一次访问时,才会调用channel的initChannel方法进行初始化，所以addHandlers方法可以放在这段之后
            bootstrap.childHandler(new ChannelInitializer<Channel>() {
              @Override
                protected void initChannel(Channel ch) throws Exception {
                  if (!requestStream.accept() || !wsStream.accept()) {
                    ch.close();
                    return;
                  }
                  ChannelPipeline pipeline = ch.pipeline();
                  //由于分析整个启动流程,这里忽略SSL的处理
                  if (sslHelper.isSSL()) {
                    ch.pipeline().addFirst("handshaker", new SslHandshakeCompletionHandler(ar -> {
                      if (ar.succeeded()) {
                        if (options.isUseAlpn()) {
                          SslHandler sslHandler = pipeline.get(SslHandler.class);
                          String protocol = sslHandler.applicationProtocol();
                          if ("h2".equals(protocol)) {
                            handleHttp2(ch);
                          } else {
                            handleHttp1(ch);
                          }
                        } else {
                          handleHttp1(ch);
                        }
                      } else {
                        HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ch.eventLoop());
                        handler.context.executeFromIO(v -> {
                          handler.handler.exceptionHandler.handle(ar.cause());
                        });
                      }
                    }));
                    if (options.isSni()) {
                      SniHandler sniHandler = new SniHandler(sslHelper.serverNameMapper(vertx));
                      pipeline.addFirst(sniHandler);
                    } else {
                      SslHandler handler = new SslHandler(sslHelper.createEngine(vertx));
                      pipeline.addFirst("ssl", handler);
                    }
                  } else {
                    if (DISABLE_H2C) {
                    //只分析这段，简化分析的难度
                      handleHttp1(ch);
                    } else {
                      IdleStateHandler idle;
                      if (options.getIdleTimeout() > 0) {
                        pipeline.addLast("idle", idle = new IdleStateHandler(0, 0, options.getIdleTimeout(), options.getIdleTimeoutUnit()));
                      } else {
                        idle = null;
                      }
                      // Handler that detects whether the HTTP/2 connection preface or just process the request
                      // with the HTTP 1.x pipeline to support H2C with prior knowledge, i.e a client that connects
                      // and uses HTTP/2 in clear text directly without an HTTP upgrade.
                      pipeline.addLast(new Http1xOrH2CHandler() {
                        @Override
                        protected void configure(ChannelHandlerContext ctx, boolean h2c) {
                          if (idle != null) {
                            // It will be re-added but this way we don't need to pay attention to order
                            pipeline.remove(idle);
                          }
                          if (h2c) {
                            handleHttp2(ctx.channel());
                          } else {
                            handleHttp1(ch);
                          }
                        }
    
                        @Override
                        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
                          if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == IdleState.ALL_IDLE) {
                            ctx.close();
                          }
                        }
    
                        @Override
                        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                          super.exceptionCaught(ctx, cause);
                          HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ctx.channel().eventLoop());
                          handler.context.executeFromIO(v -> handler.handler.exceptionHandler.handle(cause));
                        }
                      });
                    }
                  }
                }
            });
    
            addHandlers(this, listenContext);
            try {
                //绑定netty服务
              bindFuture = AsyncResolveConnectHelper.doBind(vertx, address, bootstrap);
              bindFuture.addListener(res -> {
                if (res.failed()) {
                  vertx.sharedHttpServers().remove(id);
                } else {
                  Channel serverChannel = res.result();
                  if (serverChannel.localAddress() instanceof InetSocketAddress) {
                    HttpServerImpl.this.actualPort = ((InetSocketAddress)serverChannel.localAddress()).getPort();
                  } else {
                    HttpServerImpl.this.actualPort = address.port();
                  }
                  serverChannelGroup.add(serverChannel);
                  VertxMetrics metrics = vertx.metricsSPI();
                  this.metrics = metrics != null ? metrics.createHttpServerMetrics(options, address) : null;
                }
              });
            } catch (final Throwable t) {
              // Make sure we send the exception back through the handler (if any)
              if (listenHandler != null) {
                vertx.runOnContext(v -> listenHandler.handle(Future.failedFuture(t)));
              } else {
                // No handler - log so user can see failure
                log.error(t);
              }
              listening = false;
              return this;
            }
            vertx.sharedHttpServers().put(id, this);
            actualServer = this;
          } else {
            // Server already exists with that host/port - we will use that
            actualServer = shared;
            this.actualPort = shared.actualPort;
            addHandlers(actualServer, listenContext);
            VertxMetrics metrics = vertx.metricsSPI();
            this.metrics = metrics != null ? metrics.createHttpServerMetrics(options, address) : null;
          }
          actualServer.bindFuture.addListener(future -> {
            if (listenHandler != null) {
              final AsyncResult<HttpServer> res;
              if (future.succeeded()) {
                res = Future.succeededFuture(HttpServerImpl.this);
              } else {
                res = Future.failedFuture(future.cause());
                listening = false;
              }
              listenContext.runOnContext((v) -> listenHandler.handle(res));
            } else if (future.failed()) {
              listening  = false;
              // No handler - log so user can see failure
              log.error(future.cause());
            }
          });
        }
        return this;
      }

handleHttp1：

private void handleHttp1(Channel ch) {
        //handlerMGr中包含了所有的workeventloop 而它的初始化则在下边的addHandlers方法处完成
        HandlerHolder<HttpHandlers> holder = httpHandlerMgr.chooseHandler(ch.eventLoop());
        if (holder == null) {
          sendServiceUnavailable(ch);
          return;
        }
        configureHttp1(ch.pipeline(), holder);
      }
      //在channelPipline中添加vertx实现的http handler
       private void configureHttp1(ChannelPipeline pipeline, HandlerHolder<HttpHandlers> holder) {
        if (logEnabled) {
          pipeline.addLast("logging", new LoggingHandler());
        }
        if (USE_FLASH_POLICY_HANDLER) {
          pipeline.addLast("flashpolicy", new FlashPolicyHandler());
        }
        pipeline.addLast("httpDecoder", new VertxHttpRequestDecoder(options));
        pipeline.addLast("httpEncoder", new VertxHttpResponseEncoder());
        if (options.isDecompressionSupported()) {
          pipeline.addLast("inflater", new HttpContentDecompressor(false));
        }
        if (options.isCompressionSupported()) {
          pipeline.addLast("deflater", new HttpChunkContentCompressor(options.getCompressionLevel()));
        }
        if (sslHelper.isSSL() || options.isCompressionSupported()) {
          // only add ChunkedWriteHandler when SSL is enabled otherwise it is not needed as FileRegion is used.
          pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());       // For large file / sendfile support
        }
        if (options.getIdleTimeout() > 0) {
          pipeline.addLast("idle", new IdleStateHandler(0, 0, options.getIdleTimeout(), options.getIdleTimeoutUnit()));
        }
        if (!DISABLE_H2C) {
          pipeline.addLast("h2c", new Http1xUpgradeToH2CHandler(this, httpHandlerMgr));
        }
        if (DISABLE_WEBSOCKETS) {
          // As a performance optimisation you can set a system property to disable websockets altogether which avoids
          // some casting and a header check
        } else {
          holder = new HandlerHolder<>(holder.context, new HttpHandlers(
            this,
            new WebSocketRequestHandler(metrics, holder.handler),
            holder.handler.wsHandler,
            holder.handler.connectionHandler,
            holder.handler.exceptionHandler));
          initializeWebsocketExtensions (pipeline);
        }
        HandlerHolder<HttpHandlers> holder2 = holder;
        VertxHandler<Http1xServerConnection> handler = VertxHandler.create(holder2.context, chctx -> {
          Http1xServerConnection conn = new Http1xServerConnection(holder2.context.owner(),
            sslHelper,
            options,
            chctx,
            holder2.context,
            serverOrigin,
            holder2.handler,
            metrics);
          if (metrics != null) {
            holder2.context.executeFromIO(v -> conn.metric(metrics.connected(conn.remoteAddress(), conn.remoteName())));
          }
          return conn;
        });
        handler.addHandler(conn -> {
          connectionMap.put(pipeline.channel(), conn);
        });
        handler.removeHandler(conn -> {
          connectionMap.remove(pipeline.channel());
        });
        pipeline.addLast("handler", handler);
    
      }

addHandlers：

private void addHandlers(HttpServerImpl server, ContextInternal context) {
        server.httpHandlerMgr.addHandler(
          new HttpHandlers(
            this,
            requestStream.handler(),
            wsStream.handler(),
            connectionHandler,
            exceptionHandler == null ? DEFAULT_EXCEPTION_HANDLER : exceptionHandler)
          , context);
      }

下面是router  
由于router继承了Handler

public interface Router extends Handler<HttpServerRequest> {

所以当请求到来时，channelpipline会按顺序调用到router的handler方法：

@Override
      public void handle(HttpServerRequest request) {
        if (log.isTraceEnabled()) log.trace("Router: " + System.identityHashCode(this) +
          " accepting request " + request.method() + " " + request.absoluteURI());
        new RoutingContextImpl(null, this, request, routes).next();
      }

next方法：

@Override
      public void next() {
        if (!iterateNext()) {
          checkHandleNoMatch();
        }
      }

iterateNext：

protected boolean iterateNext() {
        boolean failed = failed();
        if (currentRoute != null) { // Handle multiple handlers inside route object
          try {
            if (!failed && currentRoute.hasNextContextHandler(this)) {
              currentRouteNextHandlerIndex.incrementAndGet();
              resetMatchFailure();
              currentRoute.handleContext(this);
              return true;
            } else if (failed && currentRoute.hasNextFailureHandler(this)) {
              currentRouteNextFailureHandlerIndex.incrementAndGet();
              currentRoute.handleFailure(this);
              return true;
            }
          } catch (Throwable t) {
            handleInHandlerRuntimeFailure(currentRoute, failed, t);
            return true;
          }
        }
        while (iter.hasNext()) { // Search for more handlers
          RouteImpl route = iter.next();
          currentRouteNextHandlerIndex.set(0);
          currentRouteNextFailureHandlerIndex.set(0);
          try {
            int matchResult = route.matches(this, mountPoint(), failed);
            if (matchResult == 0) {
              if (log.isTraceEnabled()) log.trace("Route matches: " + route);
              resetMatchFailure();
              try {
                currentRoute = route;
                if (log.isTraceEnabled()) log.trace("Calling the " + (failed ? "failure" : "") + " handler");
                if (failed && currentRoute.hasNextFailureHandler(this)) {
                  currentRouteNextFailureHandlerIndex.incrementAndGet();
                  route.handleFailure(this);
                } else if (currentRoute.hasNextContextHandler(this)) {
                  currentRouteNextHandlerIndex.incrementAndGet();
                  route.handleContext(this);
                } else {
                  continue;
                }
              } catch (Throwable t) {
                handleInHandlerRuntimeFailure(route, failed, t);
              }
              return true;
            } else if (matchResult != 404) {
              this.matchFailure = matchResult;
            }
          } catch (Throwable e) {
            if (log.isTraceEnabled()) log.trace("IllegalArgumentException thrown during iteration", e);
            // Failure in matches algorithm (If the exception is instanceof IllegalArgumentException probably is a QueryStringDecoder error!)
            if (!this.response().ended())
              unhandledFailure((e instanceof IllegalArgumentException) ? 400 : -1, e, route.router());
            return true;
          }
        }
        return false;
      }

handleContext：

void handleContext(RoutingContextImplBase context) {
        Handler<RoutingContext> contextHandler;
    
        synchronized (this) {
          contextHandler = contextHandlers.get(context.currentRouteNextHandlerIndex() - 1);
        }
    
        contextHandler.handle(context);
      }