Java 类org.apache.thrift.transport.TServerTransport 实例源码

@Bean(name = "pool-server")
public TServer poolServer() throws Exception {
    TServerTransport transport = new TServerSocket(this.port());

    TThreadPoolServer.Args args = new TThreadPoolServer.Args(transport);
    args.transportFactory(new TTransportFactory());
    args.protocolFactory(new TBinaryProtocol.Factory());

    args.processor(this.processor());
    args.executorService(new ThreadPoolExecutor(env.getProperty(
            "rpc.server.min.worker.threads", Integer.class, 512), env
            .getProperty("rpc.server.max.worker.threads", Integer.class,
                    65535), env.getProperty(
            "rpc.server.thread.keep.alive.time", Long.class, 600l),
            TimeUnit.SECONDS, new SynchronousQueue<Runnable>()));

    return new TThreadPoolServer(args);
}

private TServerTransport getSSLServerTransport() {
  try {
    TServerTransport transport;
    TSSLTransportFactory.TSSLTransportParameters params =
            new TSSLTransportFactory.TSSLTransportParameters();

    params.setKeyStore(keystore, keystorePassword, getkeyManagerAlgorithm(), keystoreType);
    transport = TSSLTransportFactory.getServerSocket(
            port, 120000, InetAddress.getByName(bindAddress), params);

    ServerSocket serverSock = ((TServerSocket) transport).getServerSocket();
    if (serverSock instanceof SSLServerSocket) {
      SSLServerSocket sslServerSock = (SSLServerSocket) serverSock;
      List<String> enabledProtocols = new ArrayList<String>();
      for (String protocol : sslServerSock.getEnabledProtocols()) {
        if (!excludeProtocols.contains(protocol)) {
          enabledProtocols.add(protocol);
        }
      }
      sslServerSock.setEnabledProtocols(enabledProtocols.toArray(new String[0]));
    }
    return transport;
  } catch (Throwable throwable) {
    throw new FlumeException("Cannot start Thrift source.", throwable);
  }
}

ServerThread() throws TTransportException {
   TMultiplexedProcessor processor = new TMultiplexedProcessor();
   for (String beanName : serviceMap.keySet()) {
      IThriftServerService serverService = (IThriftServerService) serviceMap.getService(beanName);
      String processorName = serverService.getName();
      TProcessor tProcessor = serverService.getProcessor(serverService);
      processor.registerProcessor(processorName, tProcessor);
      logger.info("Register a processorName {} processorImpl {}", processorName, tProcessor);
   }

   logger.info("init default TServerTransport in addr {} port {}", applicationProperties.getAddr(), applicationProperties.getPort());
   TServerTransport tServerTransport = new TServerSocket(new InetSocketAddress(applicationProperties.getAddr(),
           applicationProperties.getPort()));
   TThreadPoolServer.Args args = new TThreadPoolServer.Args(tServerTransport);
   args.processor(processor);
   args.protocolFactory(tProtocolFactory);
   server = new TThreadPoolServer(args);
}

private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory,
                                            TProcessor processor,
                                            TTransportFactory transportFactory,
                                            int workerThreads,
                                            InetSocketAddress inetSocketAddress,
                                            int backlog,
                                            int clientTimeout)
    throws TTransportException {
  TServerTransport serverTransport = new TServerSocket(
                                         new TServerSocket.ServerSocketTransportArgs().
                                             bindAddr(inetSocketAddress).backlog(backlog).
                                             clientTimeout(clientTimeout));
  log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
  TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  if (workerThreads > 0) {
    serverArgs.maxWorkerThreads(workerThreads);
  }
  return new TThreadPoolServer(serverArgs);
}

/**
 * Server initialization.
 *
 * @throws Exception error
 */
public void start() throws Exception {
    log.info("initializing thrift server {}", getServerName());
    final ThreadFactory threadFactory = new ThreadFactoryBuilder()
        .setNameFormat(threadPoolNameFormat)
        .setUncaughtExceptionHandler((t, e) -> log.error("Uncaught exception in thread: {}", t.getName(), e))
        .build();
    final ExecutorService executorService = new ThreadPoolExecutor(
        Math.min(2, config.getThriftServerMaxWorkerThreads()),
        config.getThriftServerMaxWorkerThreads(),
        60L,
        TimeUnit.SECONDS,
        new SynchronousQueue<>(),
        threadFactory
    );
    RegistryUtil.registerThreadPool(registry, threadPoolNameFormat, (ThreadPoolExecutor) executorService);
    final int timeout = config.getThriftServerSocketClientTimeoutInSeconds() * 1000;
    final TServerTransport serverTransport = new TServerSocket(portNumber, timeout);
    startServing(executorService, serverTransport);
}

public static void main(String[] args) {
    try {
        userProfileServerHandler = new UserProfileServerHandler();
        userProfileProcessor = new UserProfileService.Processor(userProfileServerHandler);

        TMultiplexedProcessor airavataServerProcessor = new TMultiplexedProcessor();

        airavataServerProcessor.registerProcessor("UserProfileService", userProfileProcessor);

        TServerTransport serverTransport = new TServerSocket(9190);

        TServer server = new TThreadPoolServer(new TThreadPoolServer.Args(serverTransport).processor(airavataServerProcessor));

        System.out.println("Starting User Profile server...");
        server.serve();

    } catch (Exception x) {
        x.printStackTrace();
    }
}

/**
 * Constrcts a server object
 */
public HadoopThriftServer(String [] args) {

  if (args.length > 0) {
    serverPort = new Integer(args[0]);
  }
  try {
    ServerSocket ssock = createServerSocket(serverPort);
    TServerTransport serverTransport = new TServerSocket(ssock);
    Iface handler = new HadoopThriftHandler("hdfs-thrift-dhruba");
    ThriftHadoopFileSystem.Processor processor = new ThriftHadoopFileSystem.Processor(handler);
    TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
    serverArgs.minWorkerThreads = 10;
    serverArgs.processor(processor);
    serverArgs.transportFactory(new TTransportFactory());
    serverArgs.protocolFactory(new TBinaryProtocol.Factory());
    server = new TThreadPoolServer(serverArgs);
    System.out.println("Starting the hadoop thrift server on port [" + serverPort + "]...");
    HadoopThriftHandler.LOG.info("Starting the hadoop thrift server on port [" +serverPort + "]...");
    System.out.flush();

  } catch (Exception x) {
    x.printStackTrace();
  }
}

public void startServer2() throws Exception
{
    AwesomeService.Processor<AwesomeService.Iface> processor = new AwesomeService.Processor<>(referenceServer);
    TServerTransport serverTransport = new TServerSocket(9090);
    TServer server = new TSimpleServer(new TSimpleServer.Args(serverTransport).processor(processor));

    ServerRunner serverRunner = new ServerRunner(server);
    Thread serverThread = new Thread(serverRunner);

    serverThread.start();
    logger.info("Started binary interface");

    joinMethods.add(() -> {
        try {
            serverThread.join();
        } catch (InterruptedException ignored) {
        }
    });
}

public static void main(String[] args) {
    try {
        handler = new CalculatorHandler();
        processor = new Calculator.Processor(handler);

        try {
            TServerTransport serverTransport = new TServerSocket(9090);
            TServer server = new TThreadPoolServer(new TThreadPoolServer.Args(serverTransport).processor(processor));

            System.out.println("Starting the  server...");
            server.serve();
        } catch (TTransportException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    } catch (Exception x) {
        x.printStackTrace();
    }
}

public static void main(String[] args) 
       throws TTransportException, UnsupportedEncodingException {
    final String msg = "Hello Thrift!\n";
    final String stop_cmd = "STOP";
    final int buf_size = 1024*8;
    byte[] buf = new byte[buf_size];
    final int port = 9090;

    TServerTransport acceptor = new TServerSocket(9090);
    acceptor.listen();
    System.out.println("[Server] listening on port: " + port);

    String input;
    do {
        TTransport trans = acceptor.accept();
        int len = trans.read(buf, 0, buf_size);
        input = new String(buf, 0, len,"UTF-8");
        System.out.println("[Server] handling request: " + input);
        trans.write(msg.getBytes());
        trans.flush();
        trans.close();
    } while (! stop_cmd.regionMatches(0, input, 0, 4)); 

    System.out.println("[Server] exiting");
    acceptor.close();
}

public static void main(String[] args) {
    int port = 9090;

    try {
        TServerTransport serverTransport = new TServerSocket(port);

        Args processor = new TThreadPoolServer.Args(serverTransport)
                .inputTransportFactory(new TFramedTransport.Factory())
                .outputTransportFactory(new TFramedTransport.Factory())
                .processor(new Processor<>(new TestThriftServiceHandler()));
        //            processor.maxWorkerThreads = 20;
        TThreadPoolServer server = new TThreadPoolServer(processor);

        System.out.println("Starting the server...");
        server.serve();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

@BeforeClass
public static void setUp() {
    int port = 9090;

    try {
        TServerTransport serverTransport = new TServerSocket(port);

        Args processor = new TThreadPoolServer.Args(serverTransport)
                .inputTransportFactory(new TFramedTransport.Factory())
                .outputTransportFactory(new TFramedTransport.Factory())
                .processor(new Processor<>(new TestThriftServiceHandler()));
        //            processor.maxWorkerThreads = 20;
        TThreadPoolServer server = new TThreadPoolServer(processor);

        logger.info("Starting test server...");
        new Thread(server::serve).start();
        Thread.sleep(1000); // waiting server init
    } catch (Exception e) {
        e.printStackTrace();
    }
}

/**
 * Creates the server socket.
 *
 * @return the t server transport
 * @throws TTransportException the t transport exception
 */
public TServerTransport createServerSocket() throws TTransportException {
  return new TServerSocket(
      new InetSocketAddress(getNodeConfig().getThriftHost(), getNodeConfig().getThriftPort())) {
    @Override
    protected TSocket acceptImpl() throws TTransportException {
      ServerSocket serverSocket = getServerSocket();
      if (serverSocket == null) {
        throw new TTransportException(
            TTransportException.NOT_OPEN,
            "No underlying server socket."
        );
      }
      try {
        Socket result = serverSocket.accept();
        TSocketWrapper result2 = new TSocketWrapper(result);
        result2.setTimeout(0);
        openedSockets.add(result2);
        return result2;
      } catch (IOException iox) {
        throw new TTransportException(iox);
      }
    }
  };
}

/**
 * Test kaa node initialization service start.
 *
 * @throws Exception the exception
 */
@Test
public void testKaaNodeInitializationServiceStart() throws Exception {
  KaaNodeInitializationService kaaNodeInitializationService = kaaNodeInitializationServiceSpy();

  TThreadPoolServer server = Mockito.mock(TThreadPoolServer.class);
  Mockito.doNothing().when(server).serve();

  Mockito.doReturn(server).when(kaaNodeInitializationService).createServer(Mockito.any(TServerTransport.class),
      Mockito.any(TMultiplexedProcessor.class));

  kaaNodeInitializationService.start();

  Mockito.verify(controlInitializationService).start();
  Mockito.verify(bootstrapInitializationService).start();
  Mockito.verify(operationsInitializationService).start();
}

/**
 * Test kaa node initialization service start with transport exception.
 *
 * @throws Exception the exception
 */
@Test
public void testKaaNodeInitializationServiceStartTransportException() throws Exception {
  KaaNodeInitializationService kaaNodeInitializationService = kaaNodeInitializationServiceSpy();

  TThreadPoolServer server = Mockito.mock(TThreadPoolServer.class);
  Mockito.doThrow(TTransportException.class).when(server).serve();

  Mockito.doReturn(server).when(kaaNodeInitializationService).createServer(Mockito.any(TServerTransport.class),
      Mockito.any(TMultiplexedProcessor.class));

  kaaNodeInitializationService.start();

  Mockito.verify(controlInitializationService).start();
  Mockito.verify(bootstrapInitializationService).start();
  Mockito.verify(operationsInitializationService).start();
}

/**
 * Test kaa node initialization service stop.
 *
 * @throws Exception the exception
 */
@Test
public void testKaaNodeInitializationServiceStop() throws Exception {
  KaaNodeInitializationService kaaNodeInitializationService = kaaNodeInitializationServiceSpy();

  TThreadPoolServer server = Mockito.mock(TThreadPoolServer.class);
  Mockito.doNothing().when(server).serve();

  Mockito.doReturn(server).when(kaaNodeInitializationService).createServer(Mockito.any(TServerTransport.class),
      Mockito.any(TMultiplexedProcessor.class));

  kaaNodeInitializationService.start();
  kaaNodeInitializationService.stop();

  Mockito.verify(controlInitializationService).start();
  Mockito.verify(bootstrapInitializationService).start();
  Mockito.verify(operationsInitializationService).start();

  Mockito.verify(controlInitializationService).stop();
  Mockito.verify(bootstrapInitializationService).stop();
  Mockito.verify(operationsInitializationService).stop();

}

public int run(String[] args) throws Exception
{
    Configuration conf = getConf();
    int port = conf.getInt("wmr.server.bind.port", 50100);

    SubmissionDatabase.connect(conf);

    JobServiceHandler service = new JobServiceHandler(new Configuration());
    JobService.Processor processor = new JobService.Processor(service);
    TServerTransport transport = new TServerSocket(port);
    TServer server = new TSimpleServer(new Args(transport).processor(processor));

    server.serve();

    return 0;
}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

private void secure(Asker.Processor processor) {
    try {
   /*
   * Use TSSLTransportParameters to setup the required SSL parameters. In this example
   * we are setting the keystore and the keystore password. Other things like algorithms,
   * cipher suites, client auth etc can be set.
   */
        TSSLTransportFactory.TSSLTransportParameters params = new TSSLTransportFactory.TSSLTransportParameters();
        // The Keystore contains the private key
        params.setKeyStore(keyStore, keyPass, null, null);

  /*
   * Use any of the TSSLTransportFactory to get a server transport with the appropriate
   * SSL configuration. You can use the default settings if properties are set in the command line.
   * Ex: -Djavax.net.ssl.keyStore=.keystore and -Djavax.net.ssl.keyStorePassword=thrift
   *
   * Note: You need not explicitly call open(). The underlying server socket is bound on return
   * from the factory class.
   */
        TServerTransport serverTransport = TSSLTransportFactory.getServerSocket(port, 0, null, params);
        TServer server = new TSimpleServer(new Args(serverTransport).processor(processor));

        // Use this for a multi threaded server
        // TServer server = new TThreadPoolServer(new TThreadPoolServer.Args(serverTransport).processor(processor));

        System.out.println("Starting the secure server...");
        server.serve();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

public ThriftTestingSource(String handlerName, int port,
                           String protocol, String keystore,
                           String keystorePassword, String keyManagerType,
                           String keystoreType) throws Exception {
  TSSLTransportFactory.TSSLTransportParameters params =
          new TSSLTransportFactory.TSSLTransportParameters();
  params.setKeyStore(keystore, keystorePassword, keyManagerType, keystoreType);

  TServerSocket serverTransport = TSSLTransportFactory.getServerSocket(
          port, 10000, InetAddress.getByName("0.0.0.0"), params);

  ThriftSourceProtocol.Iface handler = getHandler(handlerName);

  Class serverClass = Class.forName("org.apache.thrift" +
          ".server.TThreadPoolServer");
  Class argsClass = Class.forName("org.apache.thrift.server" +
          ".TThreadPoolServer$Args");
  TServer.AbstractServerArgs args = (TServer.AbstractServerArgs) argsClass
          .getConstructor(TServerTransport.class)
          .newInstance(serverTransport);
  Method m = argsClass.getDeclaredMethod("maxWorkerThreads", int.class);
  m.invoke(args, Integer.MAX_VALUE);
  TProtocolFactory transportProtocolFactory = null;
  if (protocol != null && protocol == ThriftRpcClient.BINARY_PROTOCOL) {
    transportProtocolFactory = new TBinaryProtocol.Factory();
  } else {
    transportProtocolFactory = new TCompactProtocol.Factory();
  }
  args.protocolFactory(transportProtocolFactory);
  args.inputTransportFactory(new TFastFramedTransport.Factory());
  args.outputTransportFactory(new TFastFramedTransport.Factory());
  args.processor(new ThriftSourceProtocol.Processor<ThriftSourceProtocol.Iface>(handler));
  server = (TServer) serverClass.getConstructor(argsClass).newInstance(args);
  Executors.newSingleThreadExecutor().submit(new Runnable() {
    @Override
    public void run() {
      server.serve();
    }
  });
}

private TServerTransport getTServerTransport() {
  try {
    return new TServerSocket(new InetSocketAddress(bindAddress, port));
  } catch (Throwable throwable) {
    throw new FlumeException("Cannot start Thrift source.", throwable);
  }
}

private TServer getTThreadPoolServer() {
  TServerTransport serverTransport;
  if (enableSsl) {
    serverTransport = getSSLServerTransport();
  } else {
    serverTransport = getTServerTransport();
  }
  TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
  serverArgs.maxWorkerThreads(maxThreads);
  populateServerParams(serverArgs);
  return new TThreadPoolServer(serverArgs);
}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

public Args(TServerTransport transport, Configuration conf) {
  super(transport);
  minWorkerThreads = conf.getInt(MIN_WORKER_THREADS_CONF_KEY,
      DEFAULT_MIN_WORKER_THREADS);
  maxWorkerThreads = conf.getInt(MAX_WORKER_THREADS_CONF_KEY,
      DEFAULT_MAX_WORKER_THREADS);
  maxQueuedRequests = conf.getInt(MAX_QUEUED_REQUESTS_CONF_KEY,
      DEFAULT_MAX_QUEUED_REQUESTS);
  threadKeepAliveTimeSec = conf.getInt(THREAD_KEEP_ALIVE_TIME_SEC_CONF_KEY,
      DEFAULT_THREAD_KEEP_ALIVE_TIME_SEC);
}

private void startServing(final ExecutorService executorService, final TServerTransport serverTransport) {
    if (!stopping.get()) {
        final TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport)
            .processor(getProcessor())
            .executorService(executorService);
        server = new TThreadPoolServer(serverArgs);
        if (hasServerEventHandler()) {
            server.setServerEventHandler(getServerEventHandler());
        }

        final String threadName = getServerName() + "-thread-#" + serverThreadCount.incrementAndGet();
        new Thread(threadName) {
            @Override
            public void run() {
                log.debug("starting serving");
                try {
                    server.serve();
                } catch (Throwable t) {
                    if (!stopping.get()) {
                        log.error("Unexpected exception in {}. This probably "
                            + "means that the worker pool was exhausted. "
                            + "Increase 'metacat.thrift.server_max_worker_threads' "
                            + "from {} or throttle the number of requests. "
                            + "This server thread is not in a bad state so starting a new one.",
                            getServerName(), config.getThriftServerMaxWorkerThreads(), t);
                        startServing(executorService, serverTransport);
                    } else {
                        log.debug("stopping serving");
                    }
                }
                log.debug("started serving");
            }
        }.start();
    }
}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

/**
 * 初始化Thrift服务
 * 
 * @param serverTransport
 *            服务传输类型
 */
@Override
protected void initServer(TServerTransport serverTransport) {
    ThriftServerConfiguration configuration = getThriftServerConfiguration();
    // 使用多线程半同步半异步方式
    TThreadedSelectorServer.Args args = new TThreadedSelectorServer.Args((TNonblockingServerSocket) serverTransport)
            .transportFactory(configuration.getTransportFactory())
            .protocolFactory(configuration.getProtocolFactory());
    if (configuration.getExecutorService() != null)
        args.executorService(configuration.getExecutorService());
    server = new TThreadedSelectorServer(
            configuration.getServerArgsAspect().TThreadedSelectorServerArgsAspect(args).processor(getProcessor()));
    if (configuration.getServerEventHandler() != null)
        server.setServerEventHandler(configuration.getServerEventHandler());
}

/**
 * 初始化Thrift服务
 * <p>
 * 启动Thrift服务之前必须要进行初始化.
 * 
 * @param serverTransport
 *            服务传输类型
 */
protected void initServer(TServerTransport serverTransport) {
    ThriftServerConfiguration configuration = getThriftServerConfiguration();
    // 默认使用TThreadPoolServer方式启动Thrift服务器,对每个连接都会单独建立一个线程.
    TThreadPoolServer.Args args = new TThreadPoolServer.Args(serverTransport)
            .transportFactory(configuration.getTransportFactory())
            .protocolFactory(configuration.getProtocolFactory());
    // 如果不设置ExecutorService,则默认使用ThreadPoolExecutor实现.
    if (configuration.getExecutorService() != null)
        args.executorService(configuration.getExecutorService());
    server = new TThreadPoolServer(
            configuration.getServerArgsAspect().TThreadPoolServerArgsAspect(args).processor(getProcessor()));
    if (configuration.getServerEventHandler() != null)
        server.setServerEventHandler(configuration.getServerEventHandler());
}

protected void initServer(TServerTransport serverTransport) {
    ThriftServerConfiguration configuration = getThriftServerConfiguration();
    server = new TSimpleServer(configuration.getServerArgsAspect().TServerArgsAspect(
            new TServer.Args(serverTransport).transportFactory(configuration.getTransportFactory())
                    .protocolFactory(configuration.getProtocolFactory()).processor(getProcessor())));
    if (configuration.getServerEventHandler() != null)
        server.setServerEventHandler(configuration.getServerEventHandler());
}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

private TServerTransport getServerTransport() {
    switch (transport) {
        case BLOCKING: return getBlockingServerTransport();
        case NON_BLOCKING: return getNonBlockingServerTransport();
        default:
            throw new AssertionError("Invalid transport type: " + transport);
    }
}

private TServerTransport getBlockingServerTransport() {
    try {
        InetAddress localhost = InetAddress.getByName("localhost");
        InetSocketAddress socketAddress = new InetSocketAddress(localhost, 0);
        TServerSocket.ServerSocketTransportArgs args = new TServerSocket.ServerSocketTransportArgs()
                .bindAddr(socketAddress);

        return new TServerSocket(args);
    } catch (Exception e) {
        throw new AssertionError(e);
    }
}

private TServerTransport getNonBlockingServerTransport() {
    try {
        InetAddress localhost = InetAddress.getByName("localhost");
        InetSocketAddress socketAddress = new InetSocketAddress(localhost, 0);

        return new TNonblockingServerSocket(socketAddress);
    } catch (Exception e) {
        throw new AssertionError(e);
    }
}

protected void init() throws Exception {
    TServerTransport serverTransport = new TServerSocket( PORT );

    TBinaryProtocol.Factory bFactory = new TBinaryProtocol.Factory();

    server = new TThreadPoolServer(
            new TThreadPoolServer.Args( serverTransport )
                    .inputProtocolFactory( bFactory )
                    .outputProtocolFactory( bFactory )
                    .inputTransportFactory( getTransportFactory() )
                    .outputTransportFactory( getTransportFactory() )
                    .processor( getProcessor() ) );

    Thread startTread = new Thread() {

        @Override
        public void run() {
            server.serve();
        }

    };

    startTread.setName( "thrift-server" );

    startTread.start();

    while( !server.isServing() ) {
        Thread.sleep( 100 );
    }

    protocol = ExtensionLoader.getExtensionLoader(Protocol.class)
            .getExtension( ThriftProtocol.NAME );

    invoker = protocol.refer( getInterface(), getUrl() );

}

private static TServer getTThreadPoolServer(TProtocolFactory protocolFactory, THBaseService.Processor processor,
    TTransportFactory transportFactory, InetSocketAddress inetSocketAddress) throws TTransportException {
  TServerTransport serverTransport = new TServerSocket(inetSocketAddress);
  log.info("starting HBase ThreadPool Thrift server on " + inetSocketAddress.toString());
  TThreadPoolServer.Args serverArgs = new TThreadPoolServer.Args(serverTransport);
  serverArgs.processor(processor);
  serverArgs.transportFactory(transportFactory);
  serverArgs.protocolFactory(protocolFactory);
  return new TThreadPoolServer(serverArgs);
}