Java 类org.jboss.netty.channel.socket.ServerSocketChannelFactory 实例源码

public static synchronized ServerSocketChannelFactory createServerChannelFactory(String name, int workerNum) {
  name = name + "-" + serverCount.incrementAndGet();
  if(LOG.isInfoEnabled()){
    LOG.info("Create " + name + " ServerSocketChannelFactory. Worker:" + workerNum);
  }
  ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
  ThreadFactory bossFactory = builder.setNameFormat(name + " Server Boss #%d").build();
  ThreadFactory workerFactory = builder.setNameFormat(name + " Server Worker #%d").build();

  NioServerBossPool bossPool =
      new NioServerBossPool(Executors.newCachedThreadPool(bossFactory), 1, ThreadNameDeterminer.CURRENT);
  NioWorkerPool workerPool =
      new NioWorkerPool(Executors.newCachedThreadPool(workerFactory), workerNum, ThreadNameDeterminer.CURRENT);

  return new NioServerSocketChannelFactory(bossPool, workerPool);
}

public static synchronized ServerSocketChannelFactory createServerChannelFactory(String name, int workerNum) {
  name = name + "-" + serverCount.incrementAndGet();
  if(LOG.isInfoEnabled()){
    LOG.info("Create " + name + " ServerSocketChannelFactory. Worker:" + workerNum);
  }
  ThreadFactoryBuilder builder = new ThreadFactoryBuilder();
  ThreadFactory bossFactory = builder.setNameFormat(name + " Server Boss #%d").build();
  ThreadFactory workerFactory = builder.setNameFormat(name + " Server Worker #%d").build();

  NioServerBossPool bossPool =
      new NioServerBossPool(Executors.newCachedThreadPool(bossFactory), 1, ThreadNameDeterminer.CURRENT);
  NioWorkerPool workerPool =
      new NioWorkerPool(Executors.newCachedThreadPool(workerFactory), workerNum, ThreadNameDeterminer.CURRENT);

  return new NioServerSocketChannelFactory(bossPool, workerPool);
}

protected HttpTunnelServerChannel(ChannelFactory factory, ChannelPipeline pipeline, ChannelSink sink, ServerSocketChannelFactory inboundFactory, ChannelGroup realConnections) {
    super(factory, pipeline, sink);

    tunnelIdPrefix = Long.toHexString(random.nextLong());
    tunnels = new ConcurrentHashMap<String, HttpTunnelAcceptedChannel>();

    config = new HttpTunnelServerChannelConfig();
    realChannel = inboundFactory.newChannel(this.createRealPipeline(realConnections));
    config.setRealChannel(realChannel);

    opened = new AtomicBoolean(true);
    bindState = new AtomicReference<BindState>(BindState.UNBOUND);

    realConnections.add(realChannel);

    Channels.fireChannelOpen(this);
}

public HttpTunnelSoakTester() {
    scheduledExecutor = Executors.newSingleThreadScheduledExecutor();
    executor = Executors.newCachedThreadPool();
    ServerSocketChannelFactory serverChannelFactory = new NioServerSocketChannelFactory(
            executor, executor);
    HttpTunnelServerChannelFactory serverTunnelFactory = new HttpTunnelServerChannelFactory(
            serverChannelFactory);

    serverBootstrap = new ServerBootstrap(serverTunnelFactory);
    serverBootstrap.setPipelineFactory(createServerPipelineFactory());

    ClientSocketChannelFactory clientChannelFactory = new NioClientSocketChannelFactory(
            executor, executor);
    HttpTunnelClientChannelFactory clientTunnelFactory = new HttpTunnelClientChannelFactory(
            clientChannelFactory);

    clientBootstrap = new ClientBootstrap(clientTunnelFactory);
    clientBootstrap.setPipelineFactory(createClientPipelineFactory());
    configureProxy();

    channels = new DefaultChannelGroup();
}

/**
 * create server<br>
 * <br>
 * @param monitoredServers
 * @param serverSocketChannelFactory
 * @return
 */
public static MonitoringServer create(Set<Server> monitoredServers, ServerSocketChannelFactory serverSocketChannelFactory) {
    synchronized (LOCK) {
        if (monitoringServer != null) {
            return monitoringServer;
        }

        // monitoring router
        Router router = createMonitoringRouter();

        // create server
        monitoringServer = new MonitoringServer();

        // set server basic info
        monitoringServer.setBasicInfos(SERVER_NAME, 10081, router);

        // set monitored server
        setMonitoredServer(monitoredServers, monitoringServer);

        // create injector
        monitoringServer.injector = Guice.createInjector(Stage.PRODUCTION, new MonitoringServerModule(monitoringServer));

        // create and set serverBootstrap
        monitoringServer.serverBootstrap = new ServerBootstrap(serverSocketChannelFactory);
        monitoringServer.serverBootstrap.setPipelineFactory(monitoringServer.injector.getInstance(ChannelPipelineFactory.class));
        monitoringServer.setNettyOptions();
        logger.info("{}-serverBootstrap option is {}", monitoringServer.serverName, monitoringServer.serverBootstrap.getOptions());

        return monitoringServer;
    }
}

/**
 * create server.<br>
 * <br>
 * @param serverName
 * @param router
 * @param modules
 * @param serverSocketChannelFactory
 * @return
 */
public static Server create(String serverName, Router router, List<AbstractModule> modules,
        ServerSocketChannelFactory serverSocketChannelFactory) {
    checkArgument(StringUtils.isNotBlank(serverName), "serverName is blank.");
    checkArgument(!(serverName.equals(MonitoringServer.SERVER_NAME)), "monitoring does not allow to use as serverName.");
    checkArgument(serverName.matches(SERVER_NAME_REGEX), "serverName allows only " + SERVER_NAME_REGEX);
    checkArgument(router != null, "router is null");

    // create server
    Server server = new Server();

    // set server basic info
    server.setBasicInfos(serverName, 10080, router);

    // create injector
    List<AbstractModule> modulesForCreate = Lists.newArrayList();
    modulesForCreate.add(new ServerModule(server));
    if (modules != null) {
        for (AbstractModule item: modules) {
            logger.info("{} regists guice module... {}", server.serverName, item.getClass().getSimpleName());
        }
        modulesForCreate.addAll(modules);
    }
    server.injector = Guice.createInjector(Stage.PRODUCTION, modulesForCreate);

    // create and set serverBootstrap
    server.serverBootstrap = new ServerBootstrap(serverSocketChannelFactory);
    server.serverBootstrap.setPipelineFactory(server.injector.getInstance(ChannelPipelineFactory.class));
    server.setNettyOptions();
    logger.info("{}-serverBootstrap option is {}", server.serverName, server.serverBootstrap.getOptions());

    return server;
}

/**
 * Initialize the local Raft server.
 * <p/>
 * Sets up the service implementation classes, creates database
 * tables and starts any thread pools necessary. Following this
 * call all service classes are <strong>fully initialized</strong>.
 * Even though various threads are started they <strong>will not</strong>
 * use or interact with the service implementation classes. Callers
 * still have exclusive access to the system.
 * <p/>
 * This method should <strong>only</strong> be called once before {@link RaftAgent#start()}.
 *
 * @throws StorageException if the persistence components cannot be initialized
 * @throws IllegalStateException if this method is called multiple times
 */
public synchronized void initialize() throws StorageException {
    checkState(!running);
    checkState(!initialized);
    checkState(setupConversion);

    // start up the snapshots subsystem
    snapshotStore.initialize();
    // check that the snapshot metadata and the filesystem agree
    // FIXME (AG): this _may_ be expensive, especially if the user never bothers to clean out snapshots!
    // FIXME (AG): warning, warning - this is upfront work - probably a very, very bad idea
    snapshotStore.reconcileSnapshots();

    // initialize the log and store
    jdbcLog.initialize();
    jdbcStore.initialize();

    // initialize the various thread pools
    nonIoExecutorService = MoreExecutors.listeningDecorator(Executors.newCachedThreadPool());
    ioExecutorService = Executors.newCachedThreadPool();
    serverBossPool = new NioServerBossPool(ioExecutorService, 1);
    clientBossPool = new NioClientBossPool(ioExecutorService, 1);
    workerPool = new NioWorkerPool(ioExecutorService, 3);

    // TODO (AG): avoid creating threads in the initialize() method
    // initialize the networking subsystem
    sharedWorkerPool = new ShareableWorkerPool<NioWorker>(workerPool);
    ServerSocketChannelFactory serverChannelFactory = new NioServerSocketChannelFactory(serverBossPool, sharedWorkerPool);
    ClientSocketChannelFactory clientChannelFactory = new NioClientSocketChannelFactory(clientBossPool, sharedWorkerPool);
    raftNetworkClient.initialize(nonIoExecutorService, serverChannelFactory, clientChannelFactory, raftAlgorithm);

    raftAlgorithm.initialize();

    initialized = true;
}

@Before
public void setUp() throws Exception {
    realChannelFactory = mockContext.mock(ServerSocketChannelFactory.class);
    factory = new HttpTunnelServerChannelFactory(realChannelFactory);
    ChannelPipeline pipeline = Channels.pipeline(new SimpleChannelHandler());
    realChannel = new FakeServerSocketChannel(factory, pipeline,
            new FakeChannelSink());
}

public static Channel createServerChannel(InetSocketAddress addr, ChannelPipelineFactory pipelineFactory) {
    // TCP socket factory
    ServerSocketChannelFactory socketFactory = new NioServerSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool());

    // HTTP socket factory
    socketFactory = new HttpTunnelServerChannelFactory(socketFactory);

    final ServerBootstrap bootstrap = new ServerBootstrap(socketFactory);
    bootstrap.setPipelineFactory(pipelineFactory);

    bootstrap.setOption("child.tcpNoDelay", true);
    bootstrap.setOption("reuseAddress", true);

    return bootstrap.bind(addr);
}

@Override
protected ServerSocketChannelFactory createSocketChannelFactory() {
    return new OioServerSocketChannelFactory(createBossExecutor(), createWorkerExecutor());
}

@Override
protected ServerSocketChannelFactory createSocketChannelFactory() {
    return new OioServerSocketChannelFactory(createBossExecutor(), createWorkerExecutor());
}

@Override
protected ServerSocketChannelFactory createSocketChannelFactory() {
    return new OioServerSocketChannelFactory(createBossExecutor(), createWorkerExecutor());
}

@Override
protected ServerSocketChannelFactory createSocketChannelFactory() {
    return new OioServerSocketChannelFactory(createBossExecutor(), createWorkerExecutor());
}

public HttpTunnelServerChannelFactory(ServerSocketChannelFactory factory) {
    this.factory = factory;

    realConnections = new DefaultChannelGroup();
}

/**
 * create server for monitoring<br>
 * <br>
 * @param monitoredServer
 * @param serverSocketChannelFactory
 * @return
 */
public static MonitoringServer create(Server monitoredServer, ServerSocketChannelFactory serverSocketChannelFactory) {
    Set<Server> monitoredServers = Sets.newHashSet();
    monitoredServers.add(monitoredServer);
    return create(monitoredServers, serverSocketChannelFactory);
}

/**
 * create server.<br>
 * <br>
 * @param serverName
 * @param router
 * @param serverSocketChannelFactory
 * @return
 */
public static Server create(String serverName, Router router, ServerSocketChannelFactory serverSocketChannelFactory) {
    return create(serverName, router, null, serverSocketChannelFactory);
}