MqttServer.java 文件源码

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();
}