io.netty.channel.ChannelInitializer#io.netty.channel.socket.SocketChannel源码实例Demo

protected void startServer() throws InterruptedException {

      ServerBootstrap b = new ServerBootstrap();
      b.group(bossGroup, workerGroup)
       .channel(NioServerSocketChannel.class)
       .handler(new LoggingHandler(LogLevel.INFO))
       .childHandler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ChannelPipeline p = ch.pipeline();
               p.addLast(new LoggingHandler(LogLevel.DEBUG));
               p.addLast(new NettySimpleMessageHandler());
               p.addLast(new NettyMasterHandler(DefaultRedisKeeperServer.this, new CommandHandlerManager(), keeperConfig.getTrafficReportIntervalMillis()));
           }
       });
      serverSocketChannel = (ServerSocketChannel) b.bind(currentKeeperMeta.getPort()).sync().channel();
  }
 

public static void main(String[] args) throws Exception {
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) {
                ChannelPipeline p = ch.pipeline();
                p.addLast(new EchoServerV4FirstHandler());
                p.addLast(new EchoServerV4SecondHandler());
            }
        });

        ChannelFuture f = b.bind(8888).sync();
        f.channel().closeFuture().sync();
    }
    finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();

    // Enable HTTPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }

    p.addLast(new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    p.addLast(new HttpContentDecompressor());

    // Uncomment the following line if you don't want to handle HttpContents.
    //p.addLast(new HttpObjectAggregator(1048576));

    p.addLast(new HttpSnoopClientHandler());
}
 

public void initialize() {
	logger.info("Initializing the connector");
	
	EventLoopGroup bossGroup = new NioEventLoopGroup(configuration.getIoWorkerThreadCount());
    EventLoopGroup workerGroup = new NioEventLoopGroup(configuration.getIoWorkerThreadCount());
    bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup)
       .channel(NioServerSocketChannel.class)
       .option(ChannelOption.SO_BACKLOG, 100)
       .childOption(ChannelOption.TCP_NODELAY, true)
       .childHandler(new ChannelInitializer<SocketChannel>() {
           @Override
           public void initChannel(SocketChannel ch) throws Exception {
               ch.pipeline().addLast(new HttpRequestDecoder(), new HttpResponseEncoder(), new HttpObjectAggregator(configuration.getMaxContentLength()), AbstractHttpConnector.this);
               addChannelHandlers(ch.pipeline());
           }
       });
}
 

@SneakyThrows
public void start(int port) {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) {
                        socketChannel.pipeline().addLast(new WriteTimeoutHandler(30), new NettyServerHandler());
                    }
                });
        ChannelFuture future = bootstrap.bind().sync();
        future.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();

    if (sslCtx != null) {
        pipeline.addLast("ssl", sslCtx.newHandler(ch.alloc()));
    }

    pipeline.addLast("codec", new HttpClientCodec());

    // Remove the following line if you don't want automatic content decompression.
    pipeline.addLast("inflater", new HttpContentDecompressor());

    // to be used since huge file transfer
    pipeline.addLast("chunkedWriter", new ChunkedWriteHandler());

    pipeline.addLast("handler", new HttpUploadClientHandler());
}
 

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline ph = ch.pipeline();

    //入参说明: 读超时时间、写超时时间、所有类型的超时时间、时间格式
    ph.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
    // 解码和编码，应和客户端一致
    //传输的协议 Protobuf
    ph.addLast(new ProtobufVarint32FrameDecoder());
    ph.addLast(new ProtobufDecoder(UserMsg.User.getDefaultInstance()));
    ph.addLast(new ProtobufVarint32LengthFieldPrepender());
    ph.addLast(new ProtobufEncoder());

    //业务逻辑实现类
    ph.addLast("nettyServerHandler", new NettyServerHandler());
}
 

private ChannelInitializer getChannelInitializer() throws NoSuchAlgorithmException {

		Mac macDummy = Mac.getInstance(controller.getStaticConf().getHmacAlgorithm());

		final NettyClientPipelineFactory nettyClientPipelineFactory = new NettyClientPipelineFactory(this, sessionTable,
				macDummy.getMacLength(), controller, rl, signatureLength);

		ChannelInitializer channelInitializer = new ChannelInitializer<SocketChannel>() {
			@Override
			public void initChannel(SocketChannel ch) throws Exception {
				ch.pipeline().addLast(nettyClientPipelineFactory.getDecoder());
				ch.pipeline().addLast(nettyClientPipelineFactory.getEncoder());
				ch.pipeline().addLast(nettyClientPipelineFactory.getHandler());

			}
		};
		return channelInitializer;
	}
 

public static void main(String[] args) throws Exception {
	String ip = "127.0.0.1";
	int port = 8080;
	// Configure SSL.
	SelfSignedCertificate ssc = new SelfSignedCertificate();
	final SslContext sslCtx = SslContext.newServerContext(
			ssc.certificate(), ssc.privateKey(), null, null,
			IdentityCipherSuiteFilter.INSTANCE,
			new ApplicationProtocolConfig(Protocol.ALPN,
					SelectorFailureBehavior.FATAL_ALERT,
					SelectedListenerFailureBehavior.FATAL_ALERT,
					SelectedProtocol.SPDY_3_1.protocolName(),
					SelectedProtocol.HTTP_1_1.protocolName()), 0, 0);

	ChannelInitializer<SocketChannel> channelInit = new ChannelInitializer<SocketChannel>() {
		@Override
		protected void initChannel(SocketChannel ch) throws Exception {
			ChannelPipeline p = ch.pipeline();
			p.addLast(sslCtx.newHandler(ch.alloc()));				
			p.addLast(new SpdyOrHttpHandler());
		}
	};
	NettyServerUtil.newHttpServerBootstrap(ip, port, channelInit);
}
 

public void run() throws Exception {
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap(); // (2)
        b.group(workerGroup)
                .channel(NioSocketChannel.class) // (3)
                .handler(new ChannelInitializer<SocketChannel>() { // (4)
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(clientHandler);

                    }
                })
                .option(ChannelOption.SO_KEEPALIVE, true); // (6)

        // Bind and start to accept incoming connections.
        ChannelFuture f = b.connect("localhost", port).sync(); // (7)

        // Wait until the server socket is closed.
        // In this example, this does not happen, but you can do that to gracefully
        // shut down your server.
        f.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
    }
}
 

@Override
protected void createConnectionInternal(NetNode node, AsyncConnectionManager connectionManager,
        AsyncCallback<AsyncConnection> ac) {
    final InetSocketAddress inetSocketAddress = node.getInetSocketAddress();
    bootstrap.connect(node.getHost(), node.getPort()).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                SocketChannel ch = (SocketChannel) future.channel();
                NettyWritableChannel writableChannel = new NettyWritableChannel(ch);
                AsyncConnection conn;
                if (connectionManager != null) {
                    conn = connectionManager.createConnection(writableChannel, false);
                } else {
                    conn = new TcpClientConnection(writableChannel, NettyNetClient.this);
                }
                ch.pipeline().addLast(new NettyNetClientHandler(NettyNetClient.this, connectionManager, conn));
                conn.setInetSocketAddress(inetSocketAddress);
                conn = NettyNetClient.this.addConnection(inetSocketAddress, conn);
                ac.setAsyncResult(conn);
            } else {
                ac.setAsyncResult(future.cause());
            }
        }
    });
}
 

/**
 * Configure the pipeline for TLS NPN negotiation to HTTP/2.
 */
private void configureSSL(SocketChannel ch) {
    final ChannelPipeline p = ch.pipeline();
    // 先通过 SSL/TLS 协商版本
    p.addLast(sslCtx.newHandler(ch.alloc()));
    // 根据版本加载不同的 ChannelHandler
    p.addLast(new ApplicationProtocolNegotiationHandler(ApplicationProtocolNames.HTTP_1_1) {
        @Override
        protected void configurePipeline(ChannelHandlerContext ctx, String protocol) throws Exception {
            if (ApplicationProtocolNames.HTTP_2.equals(protocol)) {
                ctx.pipeline().addLast(bizGroup, "Http2ChannelHandler",
                    new Http2ChannelHandlerBuilder(serverHandler).build());
                return;
            }

            if (ApplicationProtocolNames.HTTP_1_1.equals(protocol)) {
                ctx.pipeline().addLast("HttpServerCodec", new HttpServerCodec());
                ctx.pipeline().addLast("HttpObjectAggregator", new HttpObjectAggregator(maxHttpContentLength));
                ctx.pipeline().addLast(bizGroup, "Http1ChannelHandler",
                    new Http1ServerChannelHandler(serverHandler));
                return;
            }

            throw new IllegalStateException("unknown protocol: " + protocol);
        }
    });
}
 

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();

    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
    }
    pipeline.addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, new ByteBuf[]{
            Unpooled.wrappedBuffer(Constant.RPC_DELIMITER.getBytes())
    }));
    pipeline.addLast(DECODER);
    pipeline.addLast(ENCODER);
    // and then unit logic.
    pipeline.addLast(new RpcClientHandler(nodeId)/*CLIENT_HANDLER*/);
    pipeline.addLast(new RpcClientDecoder());
    pipeline.addLast(new StreamRpcClientHandler());
    pipeline.addLast(new RpcClientUnitHandler());
}
 

@Override
    protected void initChannel(SocketChannel e) throws Exception {

        //protobuf处理器
//        e.pipeline().addLast("frameDecoder", new ProtobufVarint32FrameDecoder());//用于Decode前解决半包和粘包问题
        //此处需要传入自定义的protobuf的defaultInstance
        // e.pipeline().addLast("protobufDecoder",new ProtobufDecoder();
        //将请求和应答消息解码为http消息
        e.pipeline().addLast("http-codec", new HttpServerCodec());
        //HttpObjectAggregator:将Http消息的多个部分合成一条完整的消息
        e.pipeline().addLast("aggregator", new HttpObjectAggregator(65536));
        //ChunkedWriteHandler:向客户端发送HTML5文件
        e.pipeline().addLast("http-chunked", new ChunkedWriteHandler());
        //在管道中添加我们自己实现的接收数据实现方法
        e.pipeline().addLast("handler", new WebSocketServerHandler());

    }
 

public void run() throws InterruptedException {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 100)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast(new DelimiterBasedFrameDecoder(MAX_FRAME_LENGTH, false, Delimiters.lineDelimiter()));
                    p.addLast(new CarbonServerHandler(bus, configuration.getCarbon().getBaseRollup()));
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
                    logger.error(cause);
                    super.exceptionCaught(ctx, cause);
                }
            });

    // Start the server.
    b.bind(configuration.getCarbon().getBind(), configuration.getCarbon().getPort()).sync();
}
 

@PostConstruct
public void start() throws Exception {

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workGroup)
            .channelFactory(NioServerSocketChannel::new)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
                    pipeline.addLast("handler", nodeChannelInBoundHandler);
                }
            })
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_SNDBUF, 2048)
            .option(ChannelOption.SO_RCVBUF, 1024);
    bootstrap.bind(goPushNodeServerConfig.getNodePort()).sync();
    log.info("Node server start successful! listening port: {}", goPushNodeServerConfig.getNodePort());
}
 

public NettyIoAcceptor(NettyIoServiceFactory factory, IoHandler handler) {
  this.factory = factory;
  this.handler = handler;
  channelGroup = new DefaultChannelGroup("sshd-acceptor-channels", GlobalEventExecutor.INSTANCE);;
  bootstrap.group(factory.eventLoopGroup)
      .channel(NioServerSocketChannel.class)
      .option(ChannelOption.SO_BACKLOG, 100)
      .handler(new LoggingHandler(LogLevel.INFO))
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline p = ch.pipeline();
          p.addLast(new NettyIoSession(NettyIoAcceptor.this, handler).adapter);
        }
      });
}
 

public static void newServerBootstrap(String host, int port, ChannelInitializer<SocketChannel> initializer){
	EventLoopGroup bossGroup = new NioEventLoopGroup(1);
       EventLoopGroup workerGroup = new NioEventLoopGroup();
       try {
           ServerBootstrap b = new ServerBootstrap();
           b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .handler(new LoggingHandler(LogLevel.INFO))
            .childHandler(initializer)
            .bind(port).sync().channel().closeFuture().sync();	        
       } catch (Exception e){   
       	e.printStackTrace();
       } 
       finally {
           bossGroup.shutdownGracefully();
           workerGroup.shutdownGracefully();
       }
}
 

@Override
public ChannelInitializer<SocketChannel> newChannelInitializer() {

    return new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                    .addLast(loggingHandler)
                    .addLast(new IdleStateHandler(30, 0, 10))
                    .addLast("decoder", new HttpRequestDecoder())
                    .addLast("http-aggregator", new HttpObjectAggregator(65536))
                    .addLast("encoder", new HttpResponseEncoder())
                    .addLast("chunk", new ChunkedWriteHandler())
                    .addLast(businessExecutors, "business-handler", new HttpServerHandler())
                    .addLast(new HeartbeatHandler())
                    .addLast(exceptionHandler);

        }
    };
}
 

public  void remove(SocketChannel socketChannel){
    for (@SuppressWarnings("rawtypes") Map.Entry entry:map.entrySet()){
        if (entry.getValue()==socketChannel){
            log.info("#############客户端下线##############");
            log.info("下线主机名为："+entry.getKey());
            Client client=clientService.selectClientByClientIP(entry.getKey().toString());
            if(client!=null)
            {
                client.setStatus(1);
                clientMapper.updateClient(client);
                //登陆失败，删除心跳map中的数据
                NettyServer.clientMap.remove(client.getClientIp());
            }
            map.remove(entry.getKey());
            socketChannel.close();
        }
    }
}
 

@Before
public void setup() throws Exception {
    Bootstrap b = new Bootstrap();
    b.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline p = ch.pipeline();
                p.addLast(new HttpClientCodec());
                p.addLast(new HttpContentDecompressor());
                p.addLast(new SimpleChannelInboundHandler<HttpObject>() {
                    @Override
                    protected void channelRead0(ChannelHandlerContext ctx, HttpObject msg) throws Exception {
                        response = prepareResponse(ctx, msg, response);
                    }
                });
            }
        });

    channel = b.connect(HOST, PORT)
        .sync()
        .channel();
}
 

private void initializeWebSocketTransport(final NewNettyMQTTHandler handler, Config conf) {
    LOG.debug("Configuring Websocket MQTT transport");
    if(!conf.websocketEnabled) {
    	return;
    }
    
    int port = conf.websocketPort;// Integer.parseInt(webSocketPortProp);

    final MoquetteIdleTimeoutHandler timeoutHandler = new MoquetteIdleTimeoutHandler();

    String host = conf.websocketHost;
    initFactory(host, port, "Websocket MQTT", new PipelineInitializer() {

        @Override
        void init(SocketChannel channel) {
            ChannelPipeline pipeline = channel.pipeline();
            pipeline.addLast(new HttpServerCodec());
            pipeline.addLast("aggregator", new HttpObjectAggregator(65536));
            pipeline.addLast("webSocketHandler",
                    new WebSocketServerProtocolHandler("/mqtt", MQTT_SUBPROTOCOL_CSV_LIST));
            pipeline.addLast("ws2bytebufDecoder", new WebSocketFrameToByteBufDecoder());
            pipeline.addLast("bytebuf2wsEncoder", new ByteBufToWebSocketFrameEncoder());
            configureMQTTPipeline(pipeline, timeoutHandler, handler);
        }
    });
}
 

public void start() throws Exception{
    NioEventLoopGroup group = new NioEventLoopGroup();
    ServerBootstrap strap = new ServerBootstrap();
    try {
        strap.group(group).channel(NioServerSocketChannel.class).localAddress(new InetSocketAddress(port))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel sc) throws Exception{
                        sc.pipeline().addLast(new EchoServerHandler());
                    }
                });
        ChannelFuture future = strap.bind().sync();
        System.out.println(EchoServer.class.getName() + " started and listen on " + future.channel().localAddress());
        future.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}
 

public NettyIoAcceptor(NettyIoServiceFactory factory, final IoHandler handler) {
  this.factory = factory;
  this.handler = handler;
  channelGroup = new DefaultChannelGroup("sshd-acceptor-channels", GlobalEventExecutor.INSTANCE);
  bootstrap.group(factory.eventLoopGroup)
      .channel(NioServerSocketChannel.class)
      .option(ChannelOption.SO_BACKLOG, 100)
      .handler(new LoggingHandler(LogLevel.INFO))
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline p = ch.pipeline();
          p.addLast(new NettyIoSession(NettyIoAcceptor.this, handler).adapter);
        }
      });
}
 

@Override
public void create() {
  Gdx.app.setLogLevel(Application.LOG_DEBUG);

  bossGroup = new NioEventLoopGroup();
  workerGroup = new NioEventLoopGroup();
  try {
    ServerBootstrap b = new ServerBootstrap()
        .group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .childHandler(new ChannelInitializer<SocketChannel>() {
          @Override
          protected void initChannel(SocketChannel ch) {
            TcpEndpoint endpoint = new TcpEndpoint(ch, Main.this);
            Main.this.endpoint = endpoint;
            ch.pipeline()
                .addFirst(connectionLimiter)
                .addLast(new ChannelInboundHandlerAdapter() {
                  @Override
                  public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                    super.channelInactive(ctx);
                    notifyChannelInactive(ctx);
                  }
                })
                .addLast(new SizePrefixedDecoder())
                .addLast(new EndpointedChannelHandler<>(ByteBuf.class, endpoint))
                ;
          }
        })
        .option(ChannelOption.SO_BACKLOG, 128)
        .childOption(ChannelOption.TCP_NODELAY, true)
        .childOption(ChannelOption.SO_KEEPALIVE, true);

    ChannelFuture f = b.bind(PORT).sync();
  } catch (Throwable t) {
    Gdx.app.error(TAG, t.getMessage(), t);
    Gdx.app.exit();
  }
}
 

@Override
protected void initChannel(final SocketChannel ch) {
    //及时发送 与 缓存发送
    Channel channel = new NettyChannel(ch, true);
    //设置payload,添加业务线程池到channel
    channel.setAttribute(Channel.PAYLOAD, url.getPositiveInt(Constants.PAYLOAD))
            .setAttribute(Channel.BIZ_THREAD_POOL, bizThreadPool, (k, v) -> v != null);
    if (sslContext != null) {
        ch.pipeline().addFirst("ssl", sslContext.newHandler(ch.alloc()));
    }
    ch.pipeline().addLast("connection", new ConnectionChannelHandler(channel, publisher) {
        @Override
        public void channelInactive(final ChannelHandlerContext ctx) throws Exception {
            removeChannel(channel);
            super.channelInactive(ctx);
            logger.info(String.format("disconnect %s", ctx.channel().remoteAddress()));
        }
    });

    if (adapter != null) {
        ch.pipeline().addLast("adapter", new ProtocolAdapterDecoder(adapter, channel));
    } else {
        AdapterContext context = new ProtocolAdapterContext(channel, ch.pipeline());
        context.bind(codec, chain);
    }

    ChannelTransport transport = function.apply(channel, url);
    channel.setAttribute(Channel.CHANNEL_TRANSPORT, transport);
    channel.setAttribute(Channel.SERVER_CHANNEL, getServerChannel());
    addChannel(channel, transport);
}
 

@Override
protected void initSocketChannel(SocketChannel ch) {
	super.initSocketChannel(ch);
	ch.pipeline().addLast(new HttpServerCodec());
	ch.pipeline().addLast(new HttpObjectAggregator(65536));
	ch.pipeline().addLast(new ChunkedWriteHandler());
	
	HttpServerHandler httpServerHandler = new HttpServerHandler(this, httpResource);
	processHttpHandler(httpServerHandler);
	
	ch.pipeline().addLast("http", httpServerHandler);
}
 

@Override
public void initChannel(SocketChannel ch) throws Exception {

    settingsHandler = new Http2SettingsHandler(ch.newPromise());
    responseHandler = new Http2ClientResponseHandler();
    
    if (sslCtx != null) {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), host, port));
        pipeline.addLast(Http2Util.getClientAPNHandler(maxContentLength, settingsHandler, responseHandler));
    }
}
 

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc()));
    }
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new WebSocketServerCompressionHandler());
    pipeline.addLast(new WebSocketServerProtocolHandler(WEBSOCKET_PATH, subProtocols, true));
    pipeline.addLast(new WebSocketRemoteServerFrameHandler());
}
 

public static void newHttpServerBootstrap(String ip, int port, ChannelInitializer<SocketChannel>  channelInitializer){
	// Configure the server.
	int numCPUs = Runtime.getRuntime().availableProcessors();
       EventLoopGroup bossGroup = new NioEventLoopGroup(numCPUs);
       EventLoopGroup workerGroup = new NioEventLoopGroup(numCPUs);
       try {        	
       	//public service processor
           ServerBootstrap publicServerBootstrap = new ServerBootstrap();            
           publicServerBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class);
           publicServerBootstrap.childOption(ChannelOption.TCP_NODELAY, false)
           .childOption(ChannelOption.SO_KEEPALIVE, false)
           .childHandler(channelInitializer); 
           
           //bind to public access host info
           Channel ch1;
           if("*".equals(ip)){
           	ch1 = publicServerBootstrap.bind(port).sync().channel();
           } else {
           	ch1 = publicServerBootstrap.bind(ip, port).sync().channel();
           }
           System.out.println(String.format("Started OK HttpServer at %s:%d", ip, port));
           ch1.config().setConnectTimeoutMillis(1800);            
           ch1.closeFuture().sync();             
           System.out.println("Shutdown...");            
       } catch (Throwable e) {
		e.printStackTrace();
		System.exit(1);
       } finally {
           bossGroup.shutdownGracefully();
           workerGroup.shutdownGracefully();
       }
}