java类org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory的实例源码

public static void main(String[] args) {
    // Server服务启动器
    ServerBootstrap bootstrap = new ServerBootstrap(
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool()));
    // 设置一个处理客户端消息和各种消息事件的类(Handler)
    bootstrap
            .setPipelineFactory(new ChannelPipelineFactory() {
                @Override
                public ChannelPipeline getPipeline()
                        throws Exception {
                    return Channels
                            .pipeline(new HelloServerHandler());
                }
            });
    // 开放8000端口供客户端访问。
    bootstrap.bind(new InetSocketAddress(8000));
}

NettyServerCnxnFactory() {
    bootstrap = new ServerBootstrap(
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool()));
    // parent channel
    bootstrap.setOption("reuseAddress", true);
    // child channels
    bootstrap.setOption("child.tcpNoDelay", true);
    /* set socket linger to off, so that socket close does not block */
    bootstrap.setOption("child.soLinger", -1);
    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline p = Channels.pipeline();
            if (secure) {
                initSSL(p);
            }
            p.addLast("servercnxnfactory", channelHandler);

            return p;
        }
    });
}

private ServerBootstrap startHttpServer(int port,
    final Token<DelegationTokenIdentifier> token, final URI url) {
  ServerBootstrap bootstrap = new ServerBootstrap(
      new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
          Executors.newCachedThreadPool()));

  bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() throws Exception {
      return Channels.pipeline(new HttpRequestDecoder(),
          new HttpChunkAggregator(65536), new HttpResponseEncoder(),
          new CredentialsLogicHandler(token, url.toString()));
    }
  });
  bootstrap.bind(new InetSocketAddress("localhost", port));
  return bootstrap;
}

@Override
public void start() throws KairosDBException
{
    // Configure the server.
    serverBootstrap = new ServerBootstrap(
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("telnet-boss-%d").build()),
                    Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("telnet-worker-%d").build())));

    // Configure the pipeline factory.
    serverBootstrap.setPipelineFactory(this);
    serverBootstrap.setOption("child.tcpNoDelay", true);
    serverBootstrap.setOption("child.keepAlive", true);
    serverBootstrap.setOption("reuseAddress", true);

    // Bind and start to accept incoming connections.
    serverBootstrap.bind(new InetSocketAddress(address, port));
}

public static void main(String[] args) {
    String hostname = "127.0.0.1";
    int port = 5044;
    if (args.length >= 2) {
        hostname = args[0];
        port = firstNonNull(Ints.tryParse(args[1]), 5044);
    }
    if (args.length >= 1) {
        port = firstNonNull(Ints.tryParse(args[1]), 5044);
    }

    final ChannelFactory factory =
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool());
    final ServerBootstrap b = new ServerBootstrap(factory);
    b.getPipeline().addLast("beats-frame-decoder", new BeatsFrameDecoder());
    b.getPipeline().addLast("beats-codec", new BeatsCodecHandler());
    b.getPipeline().addLast("logging", new LoggingHandler());
    System.out.println("Starting listener on " + hostname + ":" + port);
    b.bind(new InetSocketAddress(hostname, port));
}

/**
 * Startup a ServerBootstrap with NioServerSocketChannelFactory using the
 * portNo specified in the constructor.
 * 
 * @return
 */
public ServerBootstrap connect() {

    bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool()));

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new MessageFrameDecoder(),  new MessageEventBagHandler(bagList));
        }
    });

    System.out.println("Binding to: localhost:" + portNo);
    bootstrap.bind(new InetSocketAddress("localhost", portNo));

    return bootstrap;

}

private ServerBootstrap connectServer(boolean simulateConflict,
        boolean simulateConflictErrorPointer) {

    ServerBootstrap bootstrap = new ServerBootstrap(
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool()));

    final MessageEventBagHandler messagEventBagHandler = new MessageEventBagHandler(
            bagList, simulateConflict, simulateConflictErrorPointer);
    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new MessageFrameDecoder(),
                    messagEventBagHandler);
        }
    });

    bootstrap.bind(new InetSocketAddress(testPort));

    return bootstrap;

}

private ServerBootstrap connectServer() {

        ServerBootstrap bootstrap = new ServerBootstrap(
                new NioServerSocketChannelFactory(
                        Executors.newCachedThreadPool(),
                        Executors.newCachedThreadPool()));

        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() throws Exception {
                return Channels.pipeline(new MessageFrameDecoder(),
                        new MessageEventBagHandler(bagList));
            }
        });

        bootstrap.bind(new InetSocketAddress(testPort));

        return bootstrap;

    }

@Override
public void connect() {

    workerService = createWorkerService(getThreadPoolType(CollectorProperties.WRITER.COLLECTOR_WORKER_THREAD_POOL));

    workerbossService = createWorkderBossService(getThreadPoolType(CollectorProperties.WRITER.COLLECTOR_WORKERBOSS_THREAD_POOL));
    channelFactory = new NioServerSocketChannelFactory(workerbossService,
            workerService);

    bootstrap = new ServerBootstrap(channelFactory);

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(ipFilterHandler,
                    new MessageFrameDecoder(), new ReadTimeoutHandler(
                            HashedWheelTimerFactory.getInstance(),
                            readTimeout, TimeUnit.MILLISECONDS),
                    metricsHandler, channelHandler);
        }
    });

    bootstrap.bind(new InetSocketAddress(port));

}

/**
 * Startup a ServerBootstrap with NioServerSocketChannelFactory using the
 * portNo specified in the constructor.
 * 
 */
private void connectLockBootstrap() {

    lockBootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newCachedThreadPool(),
            Executors.newCachedThreadPool()));

    lockBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new MessageFrameDecoder(),
                    metricHandler, lockHandler);
        }
    });

    lockBootstrap.bind(new InetSocketAddress(lockPort));

}

/**
 * Startup a ServerBootstrap with NioServerSocketChannelFactory using the
 * portNo specified in the constructor.
 * 
 */
private void connectUnlockBootstrap() {

    unlockBootstrap = new ServerBootstrap(
            new NioServerSocketChannelFactory(
                    Executors.newCachedThreadPool(),
                    Executors.newCachedThreadPool()));

    unlockBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new MessageFrameDecoder(),
                    unlockHandler);
        }
    });

    unlockBootstrap.bind(new InetSocketAddress(releaseLockPort));

}

private ServerBootstrap startHttpServer(int port,
    final Token<DelegationTokenIdentifier> token, final URI url) {
  ServerBootstrap bootstrap = new ServerBootstrap(
      new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
          Executors.newCachedThreadPool()));

  bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() throws Exception {
      return Channels.pipeline(new HttpRequestDecoder(),
          new HttpChunkAggregator(65536), new HttpResponseEncoder(),
          new CredentialsLogicHandler(token, url.toString()));
    }
  });
  bootstrap.bind(new InetSocketAddress("localhost", port));
  return bootstrap;
}

@Inject
public TSOChannelHandler(TSOServerConfig config, RequestProcessor requestProcessor, MetricsRegistry metrics) {

    this.config = config;
    this.metrics = metrics;
    this.requestProcessor = requestProcessor;
    // Setup netty listener
    this.factory = new NioServerSocketChannelFactory(
            Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("boss-%d").build()),
            Executors.newCachedThreadPool(new ThreadFactoryBuilder().setNameFormat("worker-%d").build()),
            (Runtime.getRuntime().availableProcessors() * 2 + 1) * 2);

    this.bootstrap = new ServerBootstrap(factory);
    bootstrap.setPipelineFactory(new TSOPipelineFactory(this));

}

@Inject
public ProgrammableTSOServer(int port) {
    // Setup netty listener
    factory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(new ThreadFactoryBuilder()
            .setNameFormat("boss-%d").build()), Executors.newCachedThreadPool(new ThreadFactoryBuilder()
            .setNameFormat("worker-%d").build()), (Runtime.getRuntime().availableProcessors() * 2 + 1) * 2);

    // Create the global ChannelGroup
    channelGroup = new DefaultChannelGroup(ProgrammableTSOServer.class.getName());

    ServerBootstrap bootstrap = new ServerBootstrap(factory);
    bootstrap.setPipelineFactory(new TSOChannelHandler.TSOPipelineFactory(this));

    // Add the parent channel to the group
    Channel channel = bootstrap.bind(new InetSocketAddress(port));
    channelGroup.add(channel);

    LOG.info("********** Dumb TSO Server running on port {} **********", port);
}

public void start()
{
    //Configure the server
    NioServerSocketChannelFactory nioFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
        Executors.newCachedThreadPool());
    ServerBootstrap bootstrap = new ServerBootstrap(nioFactory);

    // Configure the pipeline factory.
    bootstrap.setPipelineFactory(new TextProtocolPipelineFactory());

    //some more options
    bootstrap.setOption("child.tcpNoDelay", true);
       bootstrap.setOption("child.keepAlive", true);

    // Bind and start to accept incoming connections.
    this.channel = bootstrap.bind(new InetSocketAddress(1504));

    System.out.println("--------------------------------------------");
       System.out.println("Netty Http Server successfully loaded on port ("+1504+").....");
       System.out.println("--------------------------------------------");
}

@Override
protected boolean doLoad() throws ModuleException {

    serverBootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool(),
            workerCount));

    serverBootstrap.setPipelineFactory(new MyChannelPipelineFactory(this));
    serverBootstrap.setOption("child.tcpNoDelay", tcpNoDelay);
    serverBootstrap.setOption("child.keepAlive", tcpKeepAlive);
    if (tcpSendBufferSize > 0) {
        serverBootstrap.setOption("child.sendBufferSize", tcpSendBufferSize);
    }
    if (tcpReceiveBufferSize > 0) {
        serverBootstrap.setOption("child.receiveBufferSize", tcpReceiveBufferSize);
    }
    serverBootstrap.setOption("reuseAddress", reuseAddress);
    serverBootstrap.setOption("child.reuseAddress", reuseAddress);

    serverChannel = serverBootstrap.bind(new InetSocketAddress(port));
    logger.debug("Bound to port [{}]", port);

    return true;
}

@Override
public void start() {
  org.jboss.netty.channel.ChannelFactory factory = new NioServerSocketChannelFactory(
          Executors.newCachedThreadPool(), Executors.newCachedThreadPool());

  ServerBootstrap serverBootstrap = new ServerBootstrap(factory);
  serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() {
      EventHandler handler = new EventHandler();
      final ChannelPipeline pipeline = Channels.pipeline(handler);
      pipeline.addFirst("decoder", new LineBasedFrameDecoder(1024));
      pipeline.addLast("encoder", new StringEncoder(Charsets.UTF_8));
      return pipeline;
    }
  });

  logger.info("OpenTSDB Source starting...");

  if (host == null) {
    nettyChannel = serverBootstrap.bind(new InetSocketAddress(port));
  } else {
    nettyChannel = serverBootstrap.bind(new InetSocketAddress(host, port));
  }
  super.start();
}

@Override
protected void doOpen() throws Throwable {
    NettyHelper.setNettyLoggerFactory();
    //设置线程池(但是线程池中的线程都是守护线程，为的就是当JVM退出时候不用考虑守护线程是否已经结束)
    ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
    ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
    ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
    bootstrap = new ServerBootstrap(channelFactory); //Netty启动类
    //定义NettyHandler(这个应该是通用的Handler，只有在服务启动的时候生效一次)
    final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
    channels = nettyHandler.getChannels();
    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() {
            NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
            ChannelPipeline pipeline = Channels.pipeline();

            pipeline.addLast("decoder", adapter.getDecoder()); //增加解码处理器
            pipeline.addLast("encoder", adapter.getEncoder()); //增加编码处理器
            pipeline.addLast("handler", nettyHandler); //增加具体操作的处理器
            return pipeline;
        }
    });
    // bind
    channel = bootstrap.bind(getBindAddress());
}

private ServerBootstrap startHttpServer(int port,
    final Token<DelegationTokenIdentifier> token, final URI url) {
  ServerBootstrap bootstrap = new ServerBootstrap(
      new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
          Executors.newCachedThreadPool()));

  bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() throws Exception {
      return Channels.pipeline(new HttpRequestDecoder(),
          new HttpChunkAggregator(65536), new HttpResponseEncoder(),
          new CredentialsLogicHandler(token, url.toString()));
    }
  });
  bootstrap.bind(new InetSocketAddress("localhost", port));
  return bootstrap;
}

public synchronized void init(Configuration conf) {
    ThreadFactory bossFactory = new ThreadFactoryBuilder()
      .setNameFormat("ShuffleHandler Netty Boss #%d")
      .build();
  ThreadFactory workerFactory = new ThreadFactoryBuilder()
      .setNameFormat("ShuffleHandler Netty Worker #%d")
    .build();

  int maximumPoolSize = conf.getInt(MAXIMUM_THREAD_POOL_SIZE,
                                    DEFAULT_MAXIMUM_THREAD_POOL_SIZE);
  try {
    workerThreadPool =
      (ThreadPoolExecutor) Executors.newCachedThreadPool(workerFactory);
    workerThreadPool.setMaximumPoolSize(maximumPoolSize);
  } catch (ClassCastException e) {
    LOG.warn("Netty worker thread pool is not of type ThreadPoolExecutor", e);
  }
  LOG.info("Netty starting up with a maximum of " + maximumPoolSize +
      " worker threads");
  channelFactory = new NioServerSocketChannelFactory(
      Executors.newCachedThreadPool(bossFactory),
      workerThreadPool, maximumPoolSize);
}

@Override
public void doOpen() throws Throwable {
    ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", false));
       ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
       int ioThread = conf.getInt(Constants.IO_THREADS,Constants.DEFAULT_IO_THREADS);
       ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, ioThread);
       bootstrap = new ServerBootstrap(channelFactory);

       final NettyHandler nettyHandler = new NettyHandler(getConf(), this);
       channels = nettyHandler.getChannels();
       bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
           public ChannelPipeline getPipeline() {
               NettyCodecAdapter adapter = new NettyCodecAdapter(conf,getCodec(), NettyServer.this);
               ChannelPipeline pipeline = Channels.pipeline();
               pipeline.addLast("decoder", adapter.getDecoder());
               pipeline.addLast("encoder", adapter.getEncoder());
               pipeline.addLast("handler", nettyHandler);
               return pipeline;
           }
       });
       // bind
       channel = bootstrap.bind(getBindAddress());
}

public MasterServer(final ChannelHandler handler){
    NioServerSocketChannelFactory channelFactory=
        new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
    bootstrap=new ServerBootstrap(channelFactory);
    pipelineFactory=new ChannelPipelineFactory(){
        private final ProtobufVarint32LengthFieldPrepender frameEncoder = new ProtobufVarint32LengthFieldPrepender();
        private final ProtobufEncoder protobufEncoder = new ProtobufEncoder();
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline p = pipeline();
            p.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
            p.addLast("protobufDecoder",new ProtobufDecoder(Protocol.SocketMessage.getDefaultInstance()));
            p.addLast("frameEncoder", frameEncoder);
            p.addLast("protobufEncoder", protobufEncoder);
            p.addLast("handler", handler);
            return p;
        }

    };
    try {
        bootstrap.setPipeline(pipelineFactory.getPipeline());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public MasterServer(final ChannelHandler handler){
    NioServerSocketChannelFactory channelFactory=
        new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
    bootstrap=new ServerBootstrap(channelFactory);
    pipelineFactory=new ChannelPipelineFactory(){
        private final ProtobufVarint32LengthFieldPrepender frameEncoder = new ProtobufVarint32LengthFieldPrepender();
        private final ProtobufEncoder protobufEncoder = new ProtobufEncoder();
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline p = pipeline();
            p.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
            p.addLast("protobufDecoder",new ProtobufDecoder(Protocol.SocketMessage.getDefaultInstance()));
            p.addLast("frameEncoder", frameEncoder);
            p.addLast("protobufEncoder", protobufEncoder);
            p.addLast("handler", handler);
            return p;
        }

    };
    try {
        bootstrap.setPipeline(pipelineFactory.getPipeline());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public void start() {
    bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newFixedThreadPool(1),
            Executors.newCachedThreadPool()
    ));

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline pipeline = new DefaultChannelPipeline();
            pipeline.addLast("ssl", new SslHandler(getSSLEngine()));
            pipeline.addLast("decoder", new LumberjackDecoder());
            pipeline.addLast("logHandler", new LogEventHandler(eventListener));
            return pipeline;
        }
    });
    bootstrap.bind(new InetSocketAddress(configuration.getIpAddress(), configuration.getPort()));
}

@Override
protected void serviceInit(Configuration conf) throws Exception {
  manageOsCache = conf.getBoolean(SHUFFLE_MANAGE_OS_CACHE,
      DEFAULT_SHUFFLE_MANAGE_OS_CACHE);

  readaheadLength = conf.getInt(SHUFFLE_READAHEAD_BYTES,
      DEFAULT_SHUFFLE_READAHEAD_BYTES);

  maxShuffleConnections = conf.getInt(MAX_SHUFFLE_CONNECTIONS, 
                                      DEFAULT_MAX_SHUFFLE_CONNECTIONS);

  ThreadFactory bossFactory = new ThreadFactoryBuilder()
    .setNameFormat("ShuffleHandler Netty Boss #%d")
    .build();
  ThreadFactory workerFactory = new ThreadFactoryBuilder()
    .setNameFormat("ShuffleHandler Netty Worker #%d")
    .build();

  selector = new NioServerSocketChannelFactory(
      Executors.newCachedThreadPool(bossFactory),
      Executors.newCachedThreadPool(workerFactory));
  super.serviceInit(new Configuration(conf));
}

public void start() {
    bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newFixedThreadPool(1),
            Executors.newCachedThreadPool()
    ));

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline pipeline = new DefaultChannelPipeline();
            if(configuration.isSslEnabled()) {
                pipeline.addLast("ssl", new SslHandler(getSSLEngine()));
            }
            pipeline.addLast("decoder", new LumberjackDecoder());
            pipeline.addLast("logHandler", new EventHandler(eventListener));
            return pipeline;
        }
    });
    bootstrap.bind(new InetSocketAddress(configuration.getIpAddress(), configuration.getPort()));
}

private ServerBootstrap startHttpServer(int port,
    final Token<DelegationTokenIdentifier> token, final URI url) {
  ServerBootstrap bootstrap = new ServerBootstrap(
      new NioServerSocketChannelFactory(Executors.newCachedThreadPool(),
          Executors.newCachedThreadPool()));

  bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
    @Override
    public ChannelPipeline getPipeline() throws Exception {
      return Channels.pipeline(new HttpRequestDecoder(),
          new HttpChunkAggregator(65536), new HttpResponseEncoder(),
          new CredentialsLogicHandler(token, url.toString()));
    }
  });
  bootstrap.bind(new InetSocketAddress("localhost", port));
  return bootstrap;
}

public synchronized void start() {
    final Executor bossPool = Executors.newCachedThreadPool();
    final Executor workerPool = Executors.newCachedThreadPool();
    bootstrap = new ServerBootstrap(new NioServerSocketChannelFactory(bossPool, workerPool));
    final ClientSocketChannelFactory clientSocketChannelFactory = new NioClientSocketChannelFactory(bossPool, workerPool);
    bootstrap.setOption("child.tcpNoDelay", true);
    allChannels = new DefaultChannelGroup("handler");

    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        @Override
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new FrontendHandler(allChannels, clientSocketChannelFactory, serverPool, statistics));
        }
    });

    log.info("Starting on port {}", port);
    acceptor = bootstrap.bind(new InetSocketAddress(port));

    if (acceptor.isBound()) {
        log.info("Server started successfully");
    }
}

public MasterServer(final ChannelHandler handler){
    NioServerSocketChannelFactory channelFactory=
        new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());
    bootstrap=new ServerBootstrap(channelFactory);
    pipelineFactory=new ChannelPipelineFactory(){
        private final ProtobufVarint32LengthFieldPrepender frameEncoder = new ProtobufVarint32LengthFieldPrepender();
        private final ProtobufEncoder protobufEncoder = new ProtobufEncoder();
        public ChannelPipeline getPipeline() throws Exception {
            ChannelPipeline p = pipeline();
            p.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
            p.addLast("protobufDecoder",new ProtobufDecoder(Protocol.SocketMessage.getDefaultInstance()));
            p.addLast("frameEncoder", frameEncoder);
            p.addLast("protobufEncoder", protobufEncoder);
            p.addLast("handler", handler);
            return p;
        }

    };
    try {
        bootstrap.setPipeline(pipelineFactory.getPipeline());
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * Start rpc server which is used to update progress.
 */
private void startRPCServer() {
    this.rpcServer = new ServerBootstrap(new NioServerSocketChannelFactory(
            Executors.newFixedThreadPool(GuaguaYarnConstants.DEFAULT_STATUS_RPC_SERVER_THREAD_COUNT),
            Executors.newCachedThreadPool(new MasterThreadFactory())));

    // Set up the pipeline factory.
    this.rpcServer.setPipelineFactory(new ChannelPipelineFactory() {
        public ChannelPipeline getPipeline() throws Exception {
            return Channels.pipeline(new ObjectEncoder(),
                    new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader())),
                    new ServerHandler());
        }
    });

    // Bind and start to accept incoming connections.
    this.rpcServer.bind(new InetSocketAddress(rpcPort));
}