vertx的HttpServer模块

Start HttpServer

1/**
2   * 启动 HttpServer
3   * multi instances 采用 synchronized防止线程安全问题
4   * addHandlers 方法是actor模式的实现(EventLoopPoolSize >= instances):
5　 *               1 instances : 1 verticle(actor) : 1 VertxThread(Eventloop)
6   */
7 public synchronized HttpServer listen(int port, String host, Handler<AsyncResult<HttpServer>> listenHandler) {
8    //是否有配置requestHandler或webscoket
9    if (requestStream.handler() == null && wsStream.handler() == null) {
10      throw new IllegalStateException("Set request or websocket handler first");
11    }
12    if (listening) {
13      throw new IllegalStateException("Already listening");
14    }
15    listenContext = vertx.getOrCreateContext(); //根据currentThread 获取Context，获取null则create
16    serverOrigin = (options.isSsl() ? "https" : "http") + "://" + host + ":" + port;//判断是否启用ssl
17    List<HttpVersion> applicationProtocols = options.getAlpnVersions();//获取协议版本,默认支持1.1和2.0
18    
19    if (listenContext.isWorkerContext()) {//是否使用　Worker Verticles ,不予许使用HTTP2.0
20      applicationProtocols = applicationProtocols.stream().filter(v -> v != HttpVersion.HTTP_2).collect(Collectors.toList());
21    }
22    sslHelper.setApplicationProtocols(applicationProtocols);//应用协议
23    
24    synchronized (vertx.sharedHttpServers()) {// 监听多个不同网络接口(ip:port) Httpserver 防止并发
25      this.actualPort = port; 
26      id = new ServerID(port, host);//生成服务id
27      HttpServerImpl shared = vertx.sharedHttpServers().get(id);
28      
29      if (shared == null || port == 0) {//　mutil instances 的情况,利用 mutli core cpu
30        /**
31          * frist instances
32          */
33        serverChannelGroup = new DefaultChannelGroup("vertx-acceptor-channels", GlobalEventExecutor.INSTANCE);
34        ServerBootstrap bootstrap = new ServerBootstrap();
35        //定义两个线程组,accept size 1, 重写的VertxEventLoopGroup
36        bootstrap.group(vertx.getAcceptorEventLoopGroup(), availableWorkers);
37        
38        applyConnectionOptions(bootstrap);//添加Connection Accept之后的附属选项
39        sslHelper.validate(vertx);//验证ssl相关参数
40        bootstrap.childHandler(new ChannelInitializer<Channel>() {
41        
42          @Override
43          /**
44            * connection accept 调度切换线程后触发
45            */
46          protected void initChannel(Channel ch) throws Exception {
47          　//限流策略,读大于写，导致内存无限扩大，最终 OOM
48            if (requestStream.isPaused() || wsStream.isPaused()) {
49              ch.close(); //超过服务承载能力,关闭连接
50              return;
51            }
52            ChannelPipeline pipeline = ch.pipeline();
53            if (sslHelper.isSSL()) {//是否启用ssl
54              io.netty.util.concurrent.Future<Channel> handshakeFuture;
55              if (options.isSni()) {//是否启用sni，单服务多证书情况
56                VertxSniHandler sniHandler = new VertxSniHandler(sslHelper, vertx);
57                pipeline.addLast(sniHandler);
58                handshakeFuture = sniHandler.handshakeFuture();
59              } else {
60                SslHandler handler = new SslHandler(sslHelper.createEngine(vertx));
61                pipeline.addLast("ssl", handler);
62                handshakeFuture = handler.handshakeFuture();
63              }
64              //侦听 TLS handshake
65              handshakeFuture.addListener(future -> {
66                if (future.isSuccess()) {// 握手成功
67                  if (options.isUseAlpn()) {//是否启用alpn，协调使用的protocol
68                    //获取使用的协议
69                    SslHandler sslHandler = pipeline.get(SslHandler.class);
70                    String protocol = sslHandler.applicationProtocol();
71                    if ("h2".equals(protocol)) {//是否是http2.0
72                      handleHttp2(ch);
73                    } else {
74                      handleHttp1(ch);
75                    }
76                  } else {
77                    handleHttp1(ch);
78                  }
79                } else {//握手失败
80                  HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ch.eventLoop());
81                  handler.context.executeFromIO(() -> handler.handler.exceptionHandler.handle(future.cause()));
82                }
83              });
84            } else {
85              //是否是启用http2,通过VM Options: -Dvertx.disableH2c 设置;默认false
86              if (DISABLE_H2C) {
87                handleHttp1(ch);
88              } else {
89                IdleStateHandler idle;
90                if (options.getIdleTimeout() > 0) {//是否定义最大空闲时间
91                  pipeline.addLast("idle", idle = new IdleStateHandler(0, 0, options.getIdleTimeout()));
92                } else {
93                  idle = null;
94                }
95                
96                /**直接使用明文的http2.0或1.1处理*/
97                pipeline.addLast(new Http1xOrH2CHandler() {
98                  @Override
99                  protected void configure(ChannelHandlerContext ctx, boolean h2c) {
100                    if (idle != null) {
101                      //移除idleHandler，重新添加，不用注意次序
102                      pipeline.remove(idle);
103                    }
104                    if (h2c) {//判断协议，如果定义idle则会重新添加 idleHandler
105                      handleHttp2(ctx.channel());
106                    } else {
107                      handleHttp1(ch);
108                    }
109                  }
110
111                  @Override
112                  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
113                    if (evt instanceof IdleStateEvent && ((IdleStateEvent) evt).state() == IdleState.ALL_IDLE) {
114                      ctx.close();
115                    }
116                  }
117
118                  @Override
119                  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
120                    super.exceptionCaught(ctx, cause);
121                    //根据eventloop选中对应的handler进行异常传播
122                    HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ctx.channel().eventLoop());
123                    handler.context.executeFromIO(() -> handler.handler.exceptionHandler.handle(cause));
124                  }
125                });
126              }
127            }
128          }
129        });
130
131        addHandlers(this, listenContext);////添加一个instaces(verticle的HttpHandlers)到httpHandlerMgr中
132        try {
133        　//listen ip:port
134          bindFuture = AsyncResolveConnectHelper.doBind(vertx, SocketAddress.inetSocketAddress(port, host), bootstrap);
135          bindFuture.addListener(res -> {
136            if (res.failed()) {
137              vertx.sharedHttpServers().remove(id);
138            } else {
139              Channel serverChannel = res.result();
140              HttpServerImpl.this.actualPort = ((InetSocketAddress) serverChannel.localAddress()).getPort();
141              serverChannelGroup.add(serverChannel);//添加当前的ServerSocketChannel
142              //初始化metrcis指标
143              VertxMetrics metrics = vertx.metricsSPI();
144              this.metrics = metrics != null ? metrics.createMetrics(this, new SocketAddressImpl(port, host), options) : null;
145            }
146          });
147        } catch (final Throwable t) {
148          if (listenHandler != null) {
149            vertx.runOnContext(v -> listenHandler.handle(Future.failedFuture(t)));
150          } else {
151            log.error(t);
152          }
153          listening = false;
154          return this;
155        }
156        vertx.sharedHttpServers().put(id, this);//启动的HttpServer服务(verticle)添加到Vertx.sharedHttpMap中
157        actualServer = this;
158      } else {//other instances
159        actualServer = shared;
160        this.actualPort = shared.actualPort;
161        //在actualServer基础上添加一个instaces(verticle的HttpHandlers)到httpHandlerMgr中
162        addHandlers(actualServer, listenContext);           
163        //初始化metrics
164        VertxMetrics metrics = vertx.metricsSPI();
165        this.metrics = metrics != null ? metrics.createMetrics(this, new SocketAddressImpl(port, host), options) : null;
166      }
167      //服务 bind 状态
168      actualServer.bindFuture.addListener(future -> {
169        if (listenHandler != null) {
170          final AsyncResult<HttpServer> res;
171          if (future.succeeded()) {
172            res = Future.succeededFuture(HttpServerImpl.this);
173          } else {
174            res = Future.failedFuture(future.cause());
175            listening = false;
176          }
177          listenContext.runOnContext((v) -> listenHandler.handle(res));//回调处理
178        } else if (future.failed()) {
179          listening = false;
180          log.error(future.cause());
181        }
182      });
183    }
184    return this;
185}

如何实现隔离(actor模型)

1/**
2  * 添加一个verticle instances handlers
3  * @param server  First Actual Server(multi instances)
4  *    mutil instances 情况下第一个instance启动成功，other instances 仅仅是
5  *    利用multi core cpu，所以以 first instances actual Server为主,后续在
6  *    Current HttpServerImpl instance 添加handlers(verticle)
7  * @param context current Thread context
8  *    multi instances 下EventLoopGroup.next 方法挑选(choose)出一个Eventloop
9  *    与Context 映射. netty EventExecutor调度DefaultEventExecutorChooserFactory类
10  *    两种实现:
11  *           ①求余取模 
12  *           ②位运算取模(2的幂)
13  *    所以防止实例数量大于EventloopGroup数量，Default : 2 * CpuCoreSensor.availableProcessors()
14  *    ,linux下以读取/proc/self/status 文件为主，而不是Runtime.getRuntime().availableProcessors()
15  */
16private void addHandlers(HttpServerImpl server, ContextImpl context) {
17    server.httpHandlerMgr.addHandler(
18      new HttpHandlers(
19        requestStream.handler(),
20        wsStream.handler(),
21        connectionHandler,
22        exceptionHandler == null ? DEFAULT_EXCEPTION_HANDLER : exceptionHandler)
23      , context);
24}
25
26
27public class HttpHandlers {
28  final Handler<HttpServerRequest> requestHandler;
29  final Handler<ServerWebSocket> wsHandler;
30  final Handler<HttpConnection> connectionHandler;
31  final Handler<Throwable> exceptionHandler;
32
33  /**
34    * @param requestHandler     Http Request Handler
35    * @param wsHandler          WebScoket Handler
36    * @param connectionHander   TCP Connection Handler
37    * @param exceptionHander    Exception Handlet
38    */
39  public HttpHandlers(
40    Handler<HttpServerRequest> requestHandler,
41    Handler<ServerWebSocket> wsHandler,
42    Handler<HttpConnection> connectionHandler,
43    Handler<Throwable> exceptionHandler) {
44    this.requestHandler = requestHandler;
45    this.wsHandler = wsHandler;
46    this.connectionHandler = connectionHandler;
47    this.exceptionHandler = exceptionHandler;
48  }
49}
50
51public class HandlerManager<T> {
52    public synchronized void addHandler(T handler, ContextImpl context) {
53        /**
54          * 添加一个eventloop(Thread)到 VertxEventLoopGroup 集合中．
55          * accept状态后的read/write事件,线程调度在VertxEventLoopGroup类的next方法，
56          * vertx重写choose策略
57          */
58        EventLoop worker = context.nettyEventLoop();
59        availableWorkers.addWorker(worker);
60        /**
61          * 添加handlers,并且绑定handler和context映射关系．
62          * 注意部署的instances size不要超过EventLoopPoolSize，
63          * 否则出现 1 EventLoop : N handler(verticle)           * 导致一个eventloop上执行 N 个verticle
64          */
65        Handlers<T> handlers = new Handlers<>();
66        Handlers<T> prev = handlerMap.putIfAbsent(worker, handlers);
67        if (prev != null) {
68          handlers = prev;
69        }
70        handlers.addHandler(new HandlerHolder<>(context, handler));
71        hasHandlers = true;
72    }
73}

Connection scheduling process:

add handler to eventloop structure:

an eventloop corresponds to a handlers
an eventloop corresponds to multiple instances verticles(HandlerHolder)

HttpServer option

1public class HttpServerOptions extends NetServerOptions {  
2  //是否启用压缩，默认false
3  private boolean compressionSupported;
4  
5  //压缩级别越高cpu负荷越大，默认gzip
6  private int compressionLevel;
7  
8  //websocket最大的 Frame 大小,默认65536
9  private int maxWebsocketFrameSize;
10  
11  //websocket 最大消息大小，默认65536*4
12  private int maxWebsocketMessageSize;
13  
14  //处理WebSocket消息的约定的子协议
15  private String websocketSubProtocols;
16  
17  //是否自动处理100-Continue,默认false
18  private boolean handle100ContinueAutomatically;
19  
20  //分段传输chunk 大小，默认8192
21  private int maxChunkSize;
22  
23  //initial line 最大长度，默认 4096
24  private int maxInitialLineLength;
25  
26  //Header 最大大小,默认 8192
27  private int maxHeaderSize;
28  
29  //http2.0最大的并发流,默认100
30  private Http2Settings initialSettings;
31  
32  //支持alpn的版本,默认Http1.1和Http2.0
33  private List<HttpVersion> alpnVersions;
34  
35  //设置连接的窗口大小,默认无限制
36  private int http2ConnectionWindowSize;
37  
38  //是否启用压缩解码
39  private boolean decompressionSupported;
40  
41  //WebSocket Masked位为true。 PerformingUnMasking将是错误的,默认为false
42  private boolean acceptUnmaskedFrames;
43  
44  //默认HttpObjectDecoder的初始缓冲区大小，默认128
45  private int decoderInitialBufferSize;
46}

备注

11.建立HttpServer,配置最大的idle时间，默认tcpkeepalive配置是false，
2　　网络故障等造成TCP挥手交互失败从而导致epoll的达到FileMax,阻止后续连接,导致
3　　服务器无法提供服务; 或者启用keepalive,依靠内核TCP模块去侦测(默认2小时一次)．
4　　可用netstat工具查看应用当前网络状况
5　　
62.启用HTTP2，使用jetty开源apln-boot jar包,对JDK版本依赖关系强，需下载对应JDK版本的apln;
7  　或者使用openssl，当前服务环境都需安装，迁移服务麻烦，但是性能稍高.
8
93.具体 Route和其它HttpServer功能在 Web 模块中, core 模块只是实现Tcp相关、TLS、 　　Choose vertcile.handlers Scheduling 和codec等．

vertx的HttpServer模块

Start HttpServer

如何实现隔离(actor模型)

Connection scheduling process:

add handler to eventloop structure:

HttpServer option

备注

评论区

相关推荐