Java 类io.netty.channel.epoll.EpollServerSocketChannel 实例源码

protected AbstractNettyServer(String serverName) {
    this.serverName = Objects.requireNonNull(serverName, "server name");
    bootstrap = new ServerBootstrap();
    if (Epoll.isAvailable()) {
        bootstrap.option(ChannelOption.SO_BACKLOG, 1024).channel(EpollServerSocketChannel.class)
                .childOption(ChannelOption.SO_LINGER, 0).childOption(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
        log.info(serverName + " epoll init");
    } else {
        bootstrap.channel(NioServerSocketChannel.class);
        log.info(serverName + " nio init");
    }
    bootstrap.group(bossGroup, workerGroup).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, true).childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    initPipeline(ch.pipeline());
                }
            });
}

public PluginGrpcServer(int port) {
    this.pluginConnections = new HashMap<>();

    PlayerEvents playerEvents = new PlayerEvents(this);

    this.server = NettyServerBuilder.forPort(port)
            .keepAliveTime(1, TimeUnit.MINUTES)
            .keepAliveTimeout(5, TimeUnit.SECONDS)

            .addService(playerEvents)

            .directExecutor()
            .channelType(EpollServerSocketChannel.class)
            .bossEventLoopGroup(new EpollEventLoopGroup())
            .workerEventLoopGroup(new EpollEventLoopGroup())
            .build();

    // demoPluginConnections();
}

public void start() throws Exception {
    UnknownPandaServer.getLogger().info("Loading protocol");
    Protocol protocol = ProtocolSpecification.getProtocol();
    protocol.load();

    UnknownPandaServer.getLogger().info("Binding UniverseServer at *::" + port + " [tcp]");
    this.channel = new ServerBootstrap()
            .group(Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup())
            .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            //.childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(new ConnectionInitializer(this))
            .localAddress("", port)
            .bind()
            .addListeners(this)
            .sync()
            .channel();
}

private void groups(ServerBootstrap b) {
    if (StandardSystemProperty.OS_NAME.value().equals("Linux")) {
        bossGroup = new EpollEventLoopGroup(1);
        workerGroup = new EpollEventLoopGroup();
        b.channel(EpollServerSocketChannel.class)
                .group(bossGroup, workerGroup)
                .option(EpollChannelOption.TCP_CORK, true);
    } else {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        b.channel(NioServerSocketChannel.class)
                .group(bossGroup, workerGroup);
    }
    b.option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_BACKLOG, 100);
    logger.info("Bootstrap configuration: " + b.toString());
}

public void run() throws Exception {
    ServerBootstrap b = new ServerBootstrap();
    try {
        if (isEpollAvailable) {
            b.group(new EpollEventLoopGroup(this.conf.getEventLoopThreadCount()))
             .channel(EpollServerSocketChannel.class);
        } else {
            b.group(new NioEventLoopGroup(this.conf.getEventLoopThreadCount()))
             .channel(NioServerSocketChannel.class);
        }
        b.childHandler(new DefaultServerInitializer(conf, context))
         .option(ChannelOption.SO_BACKLOG, conf.getBacklog())
         .option(ChannelOption.SO_REUSEADDR, true);

        Channel ch = b.bind(conf.getPort()).sync().channel();
        ch.closeFuture().sync();
    } finally {

    }
}

@Override
public void init(final InetAddress address, final int port, final boolean useEpoll)
{
    final Class<? extends ServerSocketChannel> socketChannelClass;
    final LazyValue<? extends EventLoopGroup> lazyInit;
    if ((Epoll.isAvailable()) && useEpoll)
    {
        socketChannelClass = EpollServerSocketChannel.class;
        lazyInit = this.epollEventLoopGroupLazyValue;
        CoreMain.debug("[Netty] Using epoll channel type");
    }
    else
    {
        socketChannelClass = NioServerSocketChannel.class;
        lazyInit = this.nioEventLoopGroupLazyValue;
        CoreMain.debug("[Netty] Using default channel type");
    }
    this.channelFuture = new ServerBootstrap().channel(socketChannelClass).childHandler(new ServerConnectionChannel(this)).group(lazyInit.get()).localAddress(address, port).bind().syncUninterruptibly();
}

@Override
public void initialise(NetworkChannelHandler channelHandler) {
    this.channelHandler = channelHandler;

    final boolean useEpoll = this.configuration.getBoolean("epoll") && Epoll.isAvailable();

    EventLoopGroup acceptGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("acceptGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("acceptGroup"));

    EventLoopGroup ioGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("ioGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("ioGroup"));

    EventLoopGroup channelGroup = useEpoll ? new EpollEventLoopGroup(this.configuration.getInt("channelGroup")) :
            new NioEventLoopGroup(this.configuration.getInt("channelGroup"));

    this.serverBootstrap = new ServerBootstrap()
            .group(acceptGroup, ioGroup)
            .channel(useEpoll ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .childHandler(new ChannelInitialiser(channelGroup, this.channelHandler, null))
            .option(ChannelOption.SO_BACKLOG, this.configuration.getInt("backlog"))
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, this.configuration.getBoolean("tcpNoDelay"))
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
}

@Override
public boolean bind() {
    ChannelFuture future = new ServerBootstrap()
            .channel(Epoll.isAvailable() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .group(group)
            .childHandler(this)
            .bind(server.getConfiguration().getRcon().getHost(), server.getConfiguration().getRcon().getPort())
            .awaitUninterruptibly();

    if (future.isSuccess()) {
        this.channel = future.channel();
        return true;
    }

    return false;
}

private void initEventLoopGroup() {
    // try Epoll first and if that does work, use nio.
    try {
        clientGroup = new EpollEventLoopGroup();
        serverGroup = new EpollEventLoopGroup();
        serverChannelClass = EpollServerSocketChannel.class;
        clientChannelClass = EpollSocketChannel.class;
        return;
    } catch (Throwable t) {
        log.warn("Failed to initialize native (epoll) transport. Reason: {}. Proceeding with nio.", t.getMessage());
    }
    clientGroup = new NioEventLoopGroup();
    serverGroup = new NioEventLoopGroup();
    serverChannelClass = NioServerSocketChannel.class;
    clientChannelClass = NioSocketChannel.class;
}

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

        ChannelFuture f = b.bind(8888).sync();

        f.channel().closeFuture().sync();
    }
    finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

public void initChannel() throws Exception {
    bossGroup = new EpollEventLoopGroup();
    workerGroup = new EpollEventLoopGroup(pushListenerWorkerNum,
            new ThreadFactoryWithName(NettyPushListener.class));
    serverBootstarp = new ServerBootstrap()
            .group(bossGroup, workerGroup)
            .channel(EpollServerSocketChannel.class)
            .option(ChannelOption.SO_TIMEOUT, sockTimout)
            .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.ALLOCATOR,
                    PooledByteBufAllocator.DEFAULT)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childHandler(pushListenerChannelInitializer);

    serverBootstarp.bind(port).sync();

    logger.info("Netty TCP Push Listener nio provider: {} with {} workers", serverBootstarp
            .getClass().getCanonicalName(), pushListenerWorkerNum);
}

@Override
    public ServerBootstrap createBootstrap() {
        bootstrap = new ServerBootstrap();
        if (isEpollAvailable) {
            this.parentGroup = new EpollEventLoopGroup();
            this.childGroup = new EpollEventLoopGroup();
            bootstrap.channel(EpollServerSocketChannel.class);
        } else {
            this.parentGroup = new NioEventLoopGroup();
            this.childGroup = new NioEventLoopGroup();
            bootstrap.channel(NioServerSocketChannel.class);
        }
        bootstrap.group(parentGroup(), childGroup());
        bootstrap.childHandler(newChannelInitializer());

        bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
//        bootstrap.option(ChannelOption.SO_REUSEADDR, true);
//        bootstrap.childOption(ChannelOption.TCP_NODELAY, true);
        bootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

        return bootstrap;
    }

private void initEventLoopGroup() {
  // try Epoll first and if that does work, use nio.
  try {
    clientGroup = new EpollEventLoopGroup(0, namedThreads("netty-messaging-event-epoll-client-%d", log));
    serverGroup = new EpollEventLoopGroup(0, namedThreads("netty-messaging-event-epoll-server-%d", log));
    serverChannelClass = EpollServerSocketChannel.class;
    clientChannelClass = EpollSocketChannel.class;
    return;
  } catch (Throwable e) {
    log.debug("Failed to initialize native (epoll) transport. "
        + "Reason: {}. Proceeding with nio.", e.getMessage());
  }
  clientGroup = new NioEventLoopGroup(0, namedThreads("netty-messaging-event-nio-client-%d", log));
  serverGroup = new NioEventLoopGroup(0, namedThreads("netty-messaging-event-nio-server-%d", log));
  serverChannelClass = NioServerSocketChannel.class;
  clientChannelClass = NioSocketChannel.class;
}

private void groups(ServerBootstrap b) {
    if (StandardSystemProperty.OS_NAME.value().equals("Linux")) {
        bossGroup = new EpollEventLoopGroup(1);
        workerGroup = new EpollEventLoopGroup();
        b.channel(EpollServerSocketChannel.class)
                .group(bossGroup, workerGroup)
                .option(EpollChannelOption.TCP_CORK, true);
    } else {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup();
        b.channel(NioServerSocketChannel.class)
                .group(bossGroup, workerGroup);
    }
    b.option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_BACKLOG, 100);
    logger.info("Bootstrap configuration: " + b.toString());
}

private synchronized ServerBootstrap createServerBootstrap(final ChannelPipelineInitializer initializer) {
    final ServerBootstrap serverBootstrap = new ServerBootstrap();
    if (Epoll.isAvailable()) {
        serverBootstrap.channel(EpollServerSocketChannel.class);
        serverBootstrap.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
    } else {
        serverBootstrap.channel(NioServerSocketChannel.class);
    }
    final ChannelHandler serverChannelHandler = BGPChannel.createServerChannelHandler(initializer);
    serverBootstrap.childHandler(serverChannelHandler);

    serverBootstrap.option(ChannelOption.SO_BACKLOG, SOCKET_BACKLOG_SIZE);
    serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    serverBootstrap.childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, WATER_MARK);

    // Make sure we are doing round-robin processing
    serverBootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(FIX_BUFFER_SIZE));

    if (serverBootstrap.config().group() == null) {
        serverBootstrap.group(this.bossGroup, this.workerGroup);
    }
    return serverBootstrap;
}

public static void addEndpoint(final SocketAddress address){
    ServerBootstrap bootstrap = new ServerBootstrap();
    if(epollSupported && !disableEpoll){
        bootstrap.channel(EpollServerSocketChannel.class);
    }else{
        bootstrap.channel(NioServerSocketChannel.class);
    }
    bootstrap.localAddress(address);
    bootstrap.childHandler(NettyChannelInitializer.INSTANCE);
    bootstrap.group(workers);
    logger.info("Opening endpoint " + address.toString());
    bootstrap.bind().syncUninterruptibly().addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future){
            endpoints.add(future.channel());
            logger.info("Endpoint " + address.toString() + " started");
        }
    });
}

private void initEventLoopGroup() {
    // try Epoll first and if that does work, use nio.
    try {
        clientGroup = new EpollEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "epollC-%d", log));
        serverGroup = new EpollEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "epollS-%d", log));
        serverChannelClass = EpollServerSocketChannel.class;
        clientChannelClass = EpollSocketChannel.class;
        return;
    } catch (Throwable e) {
        log.debug("Failed to initialize native (epoll) transport. "
                + "Reason: {}. Proceeding with nio.", e.getMessage());
    }
    clientGroup = new NioEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "nioC-%d", log));
    serverGroup = new NioEventLoopGroup(0, groupedThreads("NettyMessagingEvt", "nioS-%d", log));
    serverChannelClass = NioServerSocketChannel.class;
    clientChannelClass = NioSocketChannel.class;
}

/** Returns the correct ServerSocketChannel class based on IOMode. */
public static Class<? extends ServerChannel> getServerChannelClass(IOMode mode) {
  switch (mode) {
    case NIO:
      return NioServerSocketChannel.class;
    case EPOLL:
      return EpollServerSocketChannel.class;
    default:
      throw new IllegalArgumentException("Unknown io mode: " + mode);
  }
}

public EpollConnDroppingServer(final int port, final int dropEveryRequest)
throws InterruptedException {
    dispatchGroup = new EpollEventLoopGroup();
    workerGroup = new EpollEventLoopGroup();
    final ServerBootstrap bootstrap = new ServerBootstrap()
        .group(dispatchGroup, workerGroup)
        .channel(EpollServerSocketChannel.class)
        .childHandler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                public final void initChannel(final SocketChannel ch) {
                    if(dropEveryRequest > 0) {
                        ch.pipeline().addLast(
                            new SimpleChannelInboundHandler<Object>() {
                                @Override
                                protected final void channelRead0(
                                    final ChannelHandlerContext ctx, final Object msg
                                ) throws Exception {
                                    if(0 == reqCounter.incrementAndGet() % dropEveryRequest) {
                                        final Channel conn = ctx.channel();
                                        System.out.println("Dropping the connection " + conn);
                                        conn.close();
                                    }
                                }
                            }
                        );
                    }
                }
            }
        );

    bindFuture = bootstrap.bind(port).sync();
}

public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollServerSocketChannel.class;
  }else{
    return NioServerSocketChannel.class;
  }
}

/** Returns the correct ServerSocketChannel class based on IOMode. */
public static Class<? extends ServerChannel> getServerChannelClass(IOMode mode) {
  switch (mode) {
    case NIO:
      return NioServerSocketChannel.class;
    case EPOLL:
      return EpollServerSocketChannel.class;
    default:
      throw new IllegalArgumentException("Unknown io mode: " + mode);
  }
}

public static void main(String... args) throws Exception {
  System.out.println("starting server");

  String host = System.getProperty("host", "0.0.0.0");
  int port = Integer.getInteger("port", 8001);
  boolean useEpoll = Boolean.getBoolean("usePoll");

  Class channel;

  if (useEpoll) {
    channel = EpollServerSocketChannel.class;
  } else  {
    channel = NioServerSocketChannel.class;
  }

  ThreadFactory tf = new DefaultThreadFactory("server-elg-", true /*daemon */);
  NioEventLoopGroup boss = new NioEventLoopGroup(1, tf);
  NioEventLoopGroup worker = new NioEventLoopGroup(0, tf);
  NettyServerBuilder builder =
      NettyServerBuilder.forPort(port)
          .bossEventLoopGroup(boss)
          .workerEventLoopGroup(worker)
          .channelType(channel)
          .addService(new DefaultService())
          .directExecutor()
          .maxConcurrentCallsPerConnection(Runtime.getRuntime().availableProcessors() * 256)
          .flowControlWindow(NettyChannelBuilder.DEFAULT_FLOW_CONTROL_WINDOW * 10);

  io.grpc.Server start = builder.build();
  start.start();

  System.out.println("server started");
  start.awaitTermination();
}

public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollServerSocketChannel.class;
  }else{
    return NioServerSocketChannel.class;
  }
}

/**
 * 初始化
 *
 * @param pThreadCount parent thread count.
 * @param cThreadCount worker thread count.
 * @param options      netty network options。
 */
public void initialize(int pThreadCount, int cThreadCount,
                       Map<ChannelOption<Object>, Object> options) {
    this.bootstrap = new ServerBootstrap();
    if (Epoll.isAvailable()) {
        this.parentGroup = new EpollEventLoopGroup(pThreadCount);
        this.childGroup = new EpollEventLoopGroup(cThreadCount);
        this.bootstrap.group(parentGroup, childGroup).channel(EpollServerSocketChannel.class);
    } else {
        this.parentGroup = new NioEventLoopGroup(pThreadCount);
        this.childGroup = new NioEventLoopGroup(cThreadCount);
        this.bootstrap.group(parentGroup, childGroup).channel(NioServerSocketChannel.class);
    }
    // handlers
    this.prepender = new LengthFieldPrepender(this.lengthFieldLength, false);
    bootstrap.childHandler(new ChannelInitializer() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ServerContext.this.initChannel(ch);
        }
    });
    //
    this.defaultOptions();
    if (!options.isEmpty()) {
        for (Map.Entry<ChannelOption<Object>, Object> entry : options.entrySet()) {
            bootstrap.childOption(entry.getKey(), entry.getValue());
        }
    }
}

private ServerBootstrap newEpoolServerBootstrap() {
  bossGroup = new EpollEventLoopGroup();
  workerGroup = new EpollEventLoopGroup();
  return new ServerBootstrap()
      .group(bossGroup, workerGroup)
      .channel(EpollServerSocketChannel.class);
}

private void executeBootstrap(ScheduledExecutor scheduledExecutor, int port, boolean useWebSocket, boolean useSsl)
        throws InterruptedException {
    ServerBootstrap bootstrap = new ServerBootstrap();

    Class<? extends ServerChannel> serverChannelClass;

    if (Literals.NETTY_EPOLL.equals(Settings.INSTANCE.nettyTransportMode())) {
        serverChannelClass = EpollServerSocketChannel.class;
    }
    else {
        serverChannelClass = NioServerSocketChannel.class;
    }

    bootstrap = bootstrap.group(bossGroup, workerGroup).channel(serverChannelClass);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);

    if (scheduledExecutor != null) {
        bootstrap.handler(scheduledExecutor);
    }

    bootstrap.childHandler(new MqttChannelInitializer(useWebSocket, useSsl));

    bootstrap.childOption(ChannelOption.TCP_NODELAY, true)
            // setting buffer size can improve I/O
            .childOption(ChannelOption.SO_SNDBUF, 1048576).childOption(ChannelOption.SO_RCVBUF, 1048576)
            // recommended in
            // http://normanmaurer.me/presentations/2014-facebook-eng-netty/slides.html#11.0
            .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(8 * 1024, 32 * 1024))
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

    bootstrap.bind(port).sync();
}

public static Class<? extends ServerSocketChannel> getServerSocketChannelClass(EventLoopGroup eventLoopGroup) {
    if (eventLoopGroup instanceof EpollEventLoopGroup) {
        return EpollServerSocketChannel.class;
    } else {
        return NioServerSocketChannel.class;
    }
}

public static Class<? extends ServerSocketChannel> getServerSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollServerSocketChannel.class;
  }else{
    return NioServerSocketChannel.class;
  }
}

private TransportTypeHolder(int workerThreads) {
    if (Epoll.isAvailable()) {
        log.info("Using native epoll transport.");
        bossGroup = new EpollEventLoopGroup(1);
        workerGroup = new EpollEventLoopGroup(workerThreads);
        channelClass = EpollServerSocketChannel.class;
    } else {
        bossGroup = new NioEventLoopGroup(1);
        workerGroup = new NioEventLoopGroup(workerThreads);
        channelClass = NioServerSocketChannel.class;
    }
}

private @NotNull ServerBootstrap serverBootstrap(@NotNull TChannel topChannel) {
    return new ServerBootstrap()
        .group(this.bossGroup, this.childGroup)
        .channel(useEpoll ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.INFO))
        .option(ChannelOption.SO_BACKLOG, 128)
        .childHandler(this.channelInitializer(true, topChannel))
        .childOption(ChannelOption.SO_KEEPALIVE, true)
        .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        .option(
            ChannelOption.WRITE_BUFFER_WATER_MARK,
            new WriteBufferWaterMark(WRITE_BUFFER_LOW_WATER_MARK, WRITE_BUFFER_HIGH_WATER_MARK)
        )
        .validate();
}

public static ServerBootstrap createServerBootstrap(
        @Nonnull final BmpSessionFactory sessionFactory,
        @Nonnull final BmpHandlerFactory hf,
        @Nonnull final BmpSessionListenerFactory slf,
        @Nonnull CreateChannel createChannel,
        @Nonnull final EventLoopGroup bossGroup,
        @Nonnull final EventLoopGroup workerGroup,
        @Nonnull final KeyMapping keys,
        boolean tryEpollSocket) {

    final ServerBootstrap serverBootstrap = new ServerBootstrap();
    serverBootstrap.childHandler(createChannel.create(sessionFactory, hf, slf));
    serverBootstrap.option(ChannelOption.SO_BACKLOG, MAX_CONNECTIONS_COUNT);
    serverBootstrap.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    serverBootstrap.group(bossGroup, workerGroup);

    if (!tryEpollSocket) {
        serverBootstrap.channel(NioServerSocketChannel.class);
    } else {
        if (Epoll.isAvailable()) {
            serverBootstrap.channel(EpollServerSocketChannel.class);
        } else {
            serverBootstrap.channel(NioServerSocketChannel.class);
        }

        if (!keys.isEmpty()) {
            if (Epoll.isAvailable()) {
                serverBootstrap.option(EpollChannelOption.TCP_MD5SIG, keys);
            } else {
                throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause());
            }
        }
    }

    return serverBootstrap;
}

synchronized ServerBootstrap createServerBootstrap(final ChannelPipelineInitializer initializer) {
    final ServerBootstrap b = new ServerBootstrap();
    b.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(final SocketChannel ch) {
            initializer.initializeChannel(ch, new DefaultPromise<>(PCEPDispatcherImpl.this.executor));
        }
    });
    b.option(ChannelOption.SO_BACKLOG, SOCKET_BACKLOG_SIZE);

    b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

    if (Epoll.isAvailable()) {
        b.channel(EpollServerSocketChannel.class);
        b.childOption(EpollChannelOption.EPOLL_MODE, EpollMode.LEVEL_TRIGGERED);
    } else {
        b.channel(NioServerSocketChannel.class);
    }
    if (!this.keys.isEmpty()) {
        if (Epoll.isAvailable()) {
            b.option(EpollChannelOption.TCP_MD5SIG, this.keys);
        } else {
            throw new UnsupportedOperationException(Epoll.unavailabilityCause().getCause());
        }
    }

    // Make sure we are doing round-robin processing
    b.childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(1));

    if (b.config().group() == null) {
        b.group(this.bossGroup, this.workerGroup);
    }

    return b;
}

private ServerBootstrap createServerBootStrap() {

        int bossThreads = Math.max(1, openFlowPorts.size());
        try {
            bossGroup = new EpollEventLoopGroup(bossThreads, groupedThreads("onos/of", "boss-%d", log));
            workerGroup = new EpollEventLoopGroup(workerThreads, groupedThreads("onos/of", "worker-%d", log));
            ServerBootstrap bs = new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(EpollServerSocketChannel.class);
            log.info("Using Epoll transport");
            return bs;
        } catch (Throwable e) {
            log.debug("Failed to initialize native (epoll) transport: {}", e.getMessage());
        }

// Requires 4.1.11 or later
//        try {
//            bossGroup = new KQueueEventLoopGroup(bossThreads, groupedThreads("onos/of", "boss-%d", log));
//            workerGroup = new KQueueEventLoopGroup(workerThreads, groupedThreads("onos/of", "worker-%d", log));
//            ServerBootstrap bs = new ServerBootstrap()
//                    .group(bossGroup, workerGroup)
//                    .channel(KQueueServerSocketChannel.class);
//            log.info("Using Kqueue transport");
//            return bs;
//        } catch (Throwable e) {
//            log.debug("Failed to initialize native (kqueue) transport. ", e.getMessage());
//        }

        bossGroup = new NioEventLoopGroup(bossThreads, groupedThreads("onos/of", "boss-%d", log));
        workerGroup = new NioEventLoopGroup(workerThreads, groupedThreads("onos/of", "worker-%d", log));
        log.info("Using Nio transport");
        return new ServerBootstrap()
                    .group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class);
    }

public void run() throws Exception {
// Configure the server.

if (Epoll.isAvailable()) {
    doRun(new EpollEventLoopGroup(), EpollServerSocketChannel.class);
} else {
    doRun(new NioEventLoopGroup(), NioServerSocketChannel.class);
}
   }

public static Class<? extends ServerSocketChannel> serverSocketChannel()
{
    return EPOLL ? EpollServerSocketChannel.class : NioServerSocketChannel.class;
}

public static Class<? extends ServerChannel> getServerChannelClass() {
    return EPOLL ? EpollServerSocketChannel.class : NioServerSocketChannel.class;
}

@Override
public void start() {
    this.defaultEventExecutorGroup = new DefaultEventExecutorGroup(
        nettyServerConfig.getServerWorkerThreads(),
        new ThreadFactory() {

            private AtomicInteger threadIndex = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "NettyServerCodecThread_" + this.threadIndex.incrementAndGet());
            }
        });

    //netty 服务端
    ServerBootstrap childHandler =
        this.serverBootstrap.group(this.eventLoopGroupBoss, this.eventLoopGroupSelector)
             //这里指定EpollServerSocketChannel或者NioServerSocketChannel类初始化channel用来接受客户端请求。
            .channel(useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class) 
            .option(ChannelOption.SO_BACKLOG, 1024)
            .option(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_KEEPALIVE, false)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_SNDBUF, nettyServerConfig.getServerSocketSndBufSize())
            .childOption(ChannelOption.SO_RCVBUF, nettyServerConfig.getServerSocketRcvBufSize())
            .localAddress(new InetSocketAddress(this.nettyServerConfig.getListenPort())) //前面有设置nettyServerConfig.setListenPort(9876)操作
            //通常会为新SocketChannel通过添加一些handler，来设置ChannelPipeline。ChannelInitializer 是一个特殊的handler，其中initChannel方法可以为SocketChannel 的pipeline添加指定handler。
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(
                        defaultEventExecutorGroup,
                        new NettyEncoder(),  //编码
                        new NettyDecoder(),  //解码
                        new IdleStateHandler(0, 0, nettyServerConfig.getServerChannelMaxIdleTimeSeconds()), //心跳检查
                        new NettyConnectManageHandler(),
                        new NettyServerHandler());
                }
            });

    if (nettyServerConfig.isServerPooledByteBufAllocatorEnable()) {
        // 这个选项有可能会占用大量堆外内存，暂时不使用。
        childHandler.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
    }

    try {
        ChannelFuture sync = this.serverBootstrap.bind().sync();//通过绑定，就可以对外提供服务了。
        InetSocketAddress addr = (InetSocketAddress) sync.channel().localAddress();
        this.port = addr.getPort();

        //奇怪为什么没有关闭？？？？ 类似这样的  sync.channel().closeFuture().sync();
    } catch (InterruptedException e1) {
        throw new RuntimeException("this.serverBootstrap.bind().sync() InterruptedException", e1);
    }

    if (this.channelEventListener != null) {
        this.nettyEventExecutor.start();
    }
    // 每隔1秒扫描下异步调用超时情况
    this.timer.scheduleAtFixedRate(new TimerTask() {

        @Override
        public void run() {
            try {
                NettyRemotingServer.this.scanResponseTable();
            } catch (Exception e) {
                log.error("scanResponseTable exception", e);
            }
        }
    }, 1000 * 3, 1000);
}

public static Class<? extends ServerChannel> serverChannelType() {
  if (Epoll.isAvailable()) {
    return EpollServerSocketChannel.class;
  }
  return NioServerSocketChannel.class;
}

/**
 * Adds a channel that listens on publicly accessible network ports
 */
public void addLanEndpoint(InetAddress address, int port) throws IOException
{
    synchronized (this.endpoints)
    {
        Class <? extends ServerSocketChannel > oclass;
        LazyLoadBase <? extends EventLoopGroup > lazyloadbase;

        if (Epoll.isAvailable() && this.mcServer.func_181035_ah())
        {
            oclass = EpollServerSocketChannel.class;
            lazyloadbase = field_181141_b;
            logger.info("Using epoll channel type");
        }
        else
        {
            oclass = NioServerSocketChannel.class;
            lazyloadbase = eventLoops;
            logger.info("Using default channel type");
        }

        this.endpoints.add(((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(oclass)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                try
                {
                    p_initChannel_1_.config().setOption(ChannelOption.TCP_NODELAY, Boolean.valueOf(true));
                }
                catch (ChannelException var3)
                {
                    ;
                }

                p_initChannel_1_.pipeline().addLast((String)"timeout", (ChannelHandler)(new ReadTimeoutHandler(30))).addLast((String)"legacy_query", (ChannelHandler)(new PingResponseHandler(NetworkSystem.this))).addLast((String)"splitter", (ChannelHandler)(new MessageDeserializer2())).addLast((String)"decoder", (ChannelHandler)(new MessageDeserializer(EnumPacketDirection.SERVERBOUND))).addLast((String)"prepender", (ChannelHandler)(new MessageSerializer2())).addLast((String)"encoder", (ChannelHandler)(new MessageSerializer(EnumPacketDirection.CLIENTBOUND)));
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
                networkmanager.setNetHandler(new NetHandlerHandshakeTCP(NetworkSystem.this.mcServer, networkmanager));
            }
        }).group((EventLoopGroup)lazyloadbase.getValue()).localAddress(address, port)).bind().syncUninterruptibly());
    }
}

/**
 * Adds a channel that listens on publicly accessible network ports
 */
public void addLanEndpoint(InetAddress address, int port) throws IOException
{
    synchronized (this.endpoints)
    {
        Class <? extends ServerSocketChannel > oclass;
        LazyLoadBase <? extends EventLoopGroup > lazyloadbase;

        if (Epoll.isAvailable() && this.mcServer.func_181035_ah())
        {
            oclass = EpollServerSocketChannel.class;
            lazyloadbase = field_181141_b;
            logger.info("Using epoll channel type");
        }
        else
        {
            oclass = NioServerSocketChannel.class;
            lazyloadbase = eventLoops;
            logger.info("Using default channel type");
        }

        this.endpoints.add(((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(oclass)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                try
                {
                    p_initChannel_1_.config().setOption(ChannelOption.TCP_NODELAY, Boolean.valueOf(true));
                }
                catch (ChannelException var3)
                {
                    ;
                }

                p_initChannel_1_.pipeline().addLast((String)"timeout", (ChannelHandler)(new ReadTimeoutHandler(30))).addLast((String)"legacy_query", (ChannelHandler)(new PingResponseHandler(NetworkSystem.this))).addLast((String)"splitter", (ChannelHandler)(new MessageDeserializer2())).addLast((String)"decoder", (ChannelHandler)(new MessageDeserializer(EnumPacketDirection.SERVERBOUND))).addLast((String)"prepender", (ChannelHandler)(new MessageSerializer2())).addLast((String)"encoder", (ChannelHandler)(new MessageSerializer(EnumPacketDirection.CLIENTBOUND)));
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
                networkmanager.setNetHandler(new NetHandlerHandshakeTCP(NetworkSystem.this.mcServer, networkmanager));
            }
        }).group((EventLoopGroup)lazyloadbase.getValue()).localAddress(address, port)).bind().syncUninterruptibly());
    }
}