vertx的HttpServer模块

/** ??* 启动 HttpServer ??* multi instances 采用 synchronized防止线程安全问题 ??* addHandlers 方法是actor模式的实现，1 instances : 1 verticle(actor) : 1 VertxThread ??*/ public synchronized HttpServer listen(int port, String host, Handler<AsyncResult<HttpServer>> listenHandler) { ???//是否有配置requestHandler或webscoket ???if (requestStream.handler() == null && wsStream.handler() == null) { ?????throw new IllegalStateException("Set request or websocket handler first"); ???} ???if (listening) { ?????throw new IllegalStateException("Already listening"); ???} ???listenContext = vertx.getOrCreateContext(); //根据currentThread 获取Context，获取null则create ???serverOrigin = (options.isSsl() ? "https" : "http") + "://" + host + ":" + port;//判断是否启用ssl ???List<HttpVersion> applicationProtocols = options.getAlpnVersions();//获取协议版本,默认支持1.1和2.0 ???????if (listenContext.isWorkerContext()) {//是否使用　Worker Verticles ,不予许使用HTTP2.0 ?????applicationProtocols = applicationProtocols.stream().filter(v -> v != HttpVersion.HTTP_2).collect(Collectors.toList()); ???} ???sslHelper.setApplicationProtocols(applicationProtocols);//应用协议 ???????synchronized (vertx.sharedHttpServers()) {//multi instances 线程安全问题 ?????this.actualPort = port; ??????id = new ServerID(port, host);//生成服务id ?????HttpServerImpl shared = vertx.sharedHttpServers().get(id); ???????????if (shared == null || port == 0) {//判断shard Map , port是否是启动过 instances ???????/** ?????????* frist instances ?????????*/ ???????serverChannelGroup = new DefaultChannelGroup("vertx-acceptor-channels", GlobalEventExecutor.INSTANCE); ???????ServerBootstrap bootstrap = new ServerBootstrap(); ???????//定义两个线程组,accept size 1, 重写的VertxEventLoopGroup ???????bootstrap.group(vertx.getAcceptorEventLoopGroup(), availableWorkers); ???????????????applyConnectionOptions(bootstrap);//添加Connection Accept之后的附属选项 ???????sslHelper.validate(vertx);//验证ssl相关参数 ???????bootstrap.childHandler(new ChannelInitializer<Channel>() { ?????????????????@Override ?????????/** ???????????* connection accept 调度切换线程后触发 ???????????*/ ?????????protected void initChannel(Channel ch) throws Exception { ?????????　//限流策略,读大于写，导致内存无限扩大，最终 OOM ???????????if (requestStream.isPaused() || wsStream.isPaused()) { ?????????????ch.close(); //超过服务承载能力,关闭连接 ?????????????return; ???????????} ???????????ChannelPipeline pipeline = ch.pipeline(); ???????????if (sslHelper.isSSL()) {//是否启用ssl ?????????????io.netty.util.concurrent.Future<Channel> handshakeFuture; ?????????????if (options.isSni()) {//是否启用sni，单服务多证书情况 ???????????????VertxSniHandler sniHandler = new VertxSniHandler(sslHelper, vertx); ???????????????pipeline.addLast(sniHandler); ???????????????handshakeFuture = sniHandler.handshakeFuture(); ?????????????} else { ???????????????SslHandler handler = new SslHandler(sslHelper.createEngine(vertx)); ???????????????pipeline.addLast("ssl", handler); ???????????????handshakeFuture = handler.handshakeFuture(); ?????????????} ?????????????//侦听 TLS handshake ?????????????handshakeFuture.addListener(future -> { ???????????????if (future.isSuccess()) {// 握手成功 ?????????????????if (options.isUseAlpn()) {//是否启用alpn，协调使用的protocol ???????????????????//获取使用的协议 ???????????????????SslHandler sslHandler = pipeline.get(SslHandler.class); ???????????????????String protocol = sslHandler.applicationProtocol(); ???????????????????if ("h2".equals(protocol)) {//是否是http2.0 ?????????????????????handleHttp2(ch); ???????????????????} else { ?????????????????????handleHttp1(ch); ???????????????????} ?????????????????} else { ???????????????????handleHttp1(ch); ?????????????????} ???????????????} else {//握手失败 ?????????????????HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ch.eventLoop()); ?????????????????handler.context.executeFromIO(() -> handler.handler.exceptionHandler.handle(future.cause())); ???????????????} ?????????????}); ???????????} else { ?????????????//是否是启用http2,通过VM Options: -Dvertx.disableH2c 设置;默认false ?????????????if (DISABLE_H2C) { ???????????????handleHttp1(ch); ?????????????} else { ???????????????IdleStateHandler idle; ???????????????if (options.getIdleTimeout() > 0) {//是否定义最大空闲时间 ?????????????????pipeline.addLast("idle", idle = new IdleStateHandler(0, 0, options.getIdleTimeout())); ???????????????} else { ?????????????????idle = null; ???????????????} ???????????????????????????????/**直接使用明文的http2.0或1.1处理*/ ???????????????pipeline.addLast(new Http1xOrH2CHandler() { ?????????????????@Override ?????????????????protected void configure(ChannelHandlerContext ctx, boolean h2c) { ???????????????????if (idle != null) { ?????????????????????//移除idleHandler，重新添加，不用注意次序 ?????????????????????pipeline.remove(idle); ???????????????????} ???????????????????if (h2c) {//判断协议，如果定义idle则会重新添加 idleHandler ?????????????????????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); ???????????????????//根据eventloop选中对应的handler进行异常传播 ???????????????????HandlerHolder<HttpHandlers> handler = httpHandlerMgr.chooseHandler(ctx.channel().eventLoop()); ???????????????????handler.context.executeFromIO(() -> handler.handler.exceptionHandler.handle(cause)); ?????????????????} ???????????????}); ?????????????} ???????????} ?????????} ???????}); ???????addHandlers(this, listenContext);//添加一个httpHandler到httpHandlerMgr中 ???????try { ???????　//listen ip:port ?????????bindFuture = AsyncResolveConnectHelper.doBind(vertx, SocketAddress.inetSocketAddress(port, host), bootstrap); ?????????bindFuture.addListener(res -> { ???????????if (res.failed()) { ?????????????vertx.sharedHttpServers().remove(id); ???????????} else { ?????????????Channel serverChannel = res.result(); ?????????????HttpServerImpl.this.actualPort = ((InetSocketAddress) serverChannel.localAddress()).getPort(); ?????????????serverChannelGroup.add(serverChannel);//添加当前的ServerSocketChannel ?????????????//初始化metrcis指标 ?????????????VertxMetrics metrics = vertx.metricsSPI(); ?????????????this.metrics = metrics != null ? metrics.createMetrics(this, new SocketAddressImpl(port, host), options) : null; ???????????} ?????????}); ???????} catch (final Throwable t) { ?????????if (listenHandler != null) { ???????????vertx.runOnContext(v -> listenHandler.handle(Future.failedFuture(t))); ?????????} else { ???????????log.error(t); ?????????} ?????????listening = false; ?????????return this; ???????} ???????vertx.sharedHttpServers().put(id, this);//启动的一个instances 添加到 Map中 ???????actualServer = this; ?????} else {//other instances ???????actualServer = shared; ???????this.actualPort = shared.actualPort; ???????addHandlers(actualServer, listenContext);//添加一个httpHandler到httpHandlerMgr中 ???????//初始化metrics ???????VertxMetrics metrics = vertx.metricsSPI(); ???????this.metrics = metrics != null ? metrics.createMetrics(this, new SocketAddressImpl(port, host), options) : null; ?????} ?????//服务 bind 状态 ?????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; ?????????log.error(future.cause()); ???????} ?????}); ???} ???return this;}

 private void addHandlers(HttpServerImpl server, ContextImpl context) { ???????server.httpHandlerMgr.addHandler( ?????????new HttpHandlers( ???????????requestStream.handler(), ???????????wsStream.handler(), ???????????connectionHandler, ???????????exceptionHandler == null ? DEFAULT_EXCEPTION_HANDLER : exceptionHandler) ?????????, context); } ???????????public class HttpHandlers { ?????final Handler<HttpServerRequest> requestHandler; ?????final Handler<ServerWebSocket> wsHandler; ?????final Handler<HttpConnection> connectionHandler; ?????final Handler<Throwable> exceptionHandler; ?????????/** ???????* @param requestHandler ????Http Request Handler ???????* @param wsHandler ?????????WebScoket Handler ???????* @param connectionHander ??TCP Connection Handler ???????* @param exceptionHander ???Exception Handlet ???????*/ ?????public HttpHandlers( ???????Handler<HttpServerRequest> requestHandler, ???????Handler<ServerWebSocket> wsHandler, ???????Handler<HttpConnection> connectionHandler, ???????Handler<Throwable> exceptionHandler) { ???????this.requestHandler = requestHandler; ???????this.wsHandler = wsHandler; ???????this.connectionHandler = connectionHandler; ???????this.exceptionHandler = exceptionHandler; ?????} ???} ???????public class HandlerManager<T> { ???????public synchronized void addHandler(T handler, ContextImpl context) { ???????????/** ?????????????* 添加一个eventloop(Thread)到 VertxEventLoopGroup 集合中． ?????????????* accept状态后的read/write事件,线程调度在VertxEventLoopGroup类的next方法， ?????????????* vertx重写choose策略 ?????????????*/ ???????????EventLoop worker = context.nettyEventLoop(); ???????????availableWorkers.addWorker(worker); ???????????/** ?????????????* 添加handlers,并且绑定handler和context映射关系． ?????????????* 注意部署的instances size不要超过EventLoopPoolSize， ?????????????* 否则出现 1 context : N handler(verticle) ?????????????*/ ???????????Handlers<T> handlers = new Handlers<>(); ???????????Handlers<T> prev = handlerMap.putIfAbsent(worker, handlers); ???????????if (prev != null) { ?????????????handlers = prev; ???????????} ???????????handlers.addHandler(new HandlerHolder<>(context, handler)); ???????????hasHandlers = true; ???????} ???}