RpcServerBootstrap.java 文件源码

private void initHttpBootstrap(int myport) {
    logger.info("initHttpBootstrap...........");
    final ServerConfig serverConfig = new ServerConfig(myport);
    final ChannelGroup channelGroup = new DefaultChannelGroup(getClass().getName());
    bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            //建议用ThreadPoolExecutor代替
            Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool(), serverConfig.getThreadCnt()));
    //设置常见参数
    bootstrap.setOption("tcpNoDelay","true");//禁用nagle算法
    bootstrap.setOption("reuseAddress", "true");
    bootstrap.setOption("SO_RCVBUF",1024*128);
    bootstrap.setOption("SO_SNDBUF",1024*128);
    timer = new HashedWheelTimer();
    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline pipeline = Channels.pipeline();
            int readTimeout = serverConfig.getReadTimeout();
            if (readTimeout > 0) {
                pipeline.addLast("timeout", new ReadTimeoutHandler(timer, readTimeout, TimeUnit.MILLISECONDS));
            }
            pipeline.addLast("decoder", new RpcRequestDecode());
            pipeline.addLast("encoder", new RpcResponseEncode());
            pipeline.addLast("handler", new NettyRpcServerHandler(channelGroup));
            return pipeline;
        }
    });

    int port = serverConfig.getPort();
    if (!checkPortConfig(port)) {
        throw new IllegalStateException("port: " + port + " already in use!");
    }

    Channel channel = bootstrap.bind(new InetSocketAddress(port));
    channelGroup.add(channel);
    logger.info("voyage server started");

    waitForShutdownCommand();
    ChannelGroupFuture future = channelGroup.close();
    future.awaitUninterruptibly();
    bootstrap.releaseExternalResources();
    timer.stop();
    timer = null;

    logger.info("voyage server stoped");

}