Java 类io.netty.channel.EventLoopGroup 实例源码

private Server(
    AddressResolver addressResolver,
    EventLoopGroup eventLoopGroup,
    Class<? extends ServerChannel> channelClass,
    boolean customEventLoop,
    Timer timer,
    boolean customTimer,
    long bindTimeoutInNanos,
    StubStore stubStore,
    boolean activityLogging) {
  this(
      addressResolver,
      eventLoopGroup,
      customEventLoop,
      timer,
      customTimer,
      bindTimeoutInNanos,
      stubStore,
      activityLogging,
      new ServerBootstrap()
          .group(eventLoopGroup)
          .channel(channelClass)
          .childHandler(new Initializer()));
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

private static Pair<EventLoopGroup, Class<? extends Channel>> createEventLoopGroup(
    Configuration conf) {
  // Max amount of threads to use. 0 lets Netty decide based on amount of cores
  int maxThreads = conf.getInt(CLIENT_MAX_THREADS, 0);

  // Config to enable native transport. Does not seem to be stable at time of implementation
  // although it is not extensively tested.
  boolean epollEnabled = conf.getBoolean(USE_NATIVE_TRANSPORT, false);

  // Use the faster native epoll transport mechanism on linux if enabled
  if (epollEnabled && JVM.isLinux() && JVM.isAmd64()) {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create EpollEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new EpollEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), EpollSocketChannel.class);
  } else {
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create NioEventLoopGroup with maxThreads = " + maxThreads);
    }
    return new Pair<EventLoopGroup, Class<? extends Channel>>(new NioEventLoopGroup(maxThreads,
        Threads.newDaemonThreadFactory("AsyncRpcChannel")), NioSocketChannel.class);
  }
}

public void start(String ip, int port) throws Exception {
    // Configure SSL.
    final SslContext sslCtx;
    if (SSL) {
        sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group).channel(NioSocketChannel.class).handler(new FileClientInitializer(sslCtx));
        Channel ch = b.connect(ip, port).sync().channel();
        ConfigurationContext.propMap.putIfAbsent(SOCKET_CHANNEL, ch);           
    }catch(Exception e){
        e.printStackTrace();
    }
}

/**
 *@description 监听指定端口
 *@time 创建时间:2017年7月21日下午3:50:26
 *@param port
 *@throws InterruptedException
 *@author dzn
 */
public void bind(int port) throws InterruptedException{
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workGroup = new NioEventLoopGroup();
    try{
        ServerBootstrap server = new ServerBootstrap();
        server.group(bossGroup, workGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .childHandler(new ChildChannelHandler());
        ChannelFuture cf = server.bind(port).sync();
        System.out.println("服务器已启动, 监控端口号为 : " + port);
        cf.channel().closeFuture().sync();
    }finally{
        bossGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }
}

@Deprecated
public RedisClient(final Timer timer, ExecutorService executor, EventLoopGroup group, Class<? extends SocketChannel> socketChannelClass, String host, int port, 
                    int connectTimeout, int commandTimeout) {
    RedisClientConfig config = new RedisClientConfig();
    config.setTimer(timer).setExecutor(executor).setGroup(group).setSocketChannelClass(socketChannelClass)
    .setAddress(host, port).setConnectTimeout(connectTimeout).setCommandTimeout(commandTimeout);

    this.config = config;
    this.executor = config.getExecutor();
    this.timer = config.getTimer();

    addr = new InetSocketAddress(config.getAddress().getHost(), config.getAddress().getPort());

    channels = new DefaultChannelGroup(config.getGroup().next());
    bootstrap = createBootstrap(config, Type.PLAIN);
    pubSubBootstrap = createBootstrap(config, Type.PUBSUB);

    this.commandTimeout = config.getCommandTimeout();
}

/**
 * Start the server
 *
 * @param port The port on which the server listen to
 */
public void run(final int port) {
    final EventLoopGroup bossGroup = new NioEventLoopGroup();
    final EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {
        final ServerBootstrap bootstrap = new ServerBootstrap()
                .group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ServerInitializer())
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.TCP_NODELAY, true);

        final ChannelFuture f = bootstrap.bind(port).sync();
        LOGGER.info("NettyServer: running on port {}", port);

        f.channel().closeFuture().sync();
    } catch (final InterruptedException e) {
        LOGGER.error("NettyServer: an error occurred while running: {}", e.getMessage());
    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

/**
 *@description 监听指定端口
 *@time 创建时间:2017年7月21日下午3:50:26
 *@param port
 *@throws InterruptedException
 *@author dzn
 */
public void bind(int port) throws InterruptedException{
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workGroup = new NioEventLoopGroup();
    try{
        ServerBootstrap server = new ServerBootstrap();
        server.group(bossGroup, workGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .childHandler(new ChildChannelHandler());
        ChannelFuture cf = server.bind(port).sync();
        System.out.println("服务器已启动, 监控端口号为 : " + port);
        cf.channel().closeFuture().sync();
    }finally{
        bossGroup.shutdownGracefully();
        workGroup.shutdownGracefully();
    }
}

public void run() {
    try {
        // Configure the server.
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.option(ChannelOption.SO_BACKLOG, 1024);
            b.group(group)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new Http2ServerInitializer(mSslCtx));

            sServerChannel = b.bind(PORT).sync().channel();
            Log.i(TAG, "Netty HTTP/2 server started on " + getServerUrl());
            sBlock.open();
            sServerChannel.closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
        Log.i(TAG, "Stopped Http2TestServerRunnable!");
    } catch (Exception e) {
        Log.e(TAG, e.toString());
    }
}

public void start() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(this.host, this.port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        System.out.println("connected server...");
                        ch.pipeline().addLast(new ByteArrayEncoder());
                        ch.pipeline().addLast(new ByteArrayDecoder());
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });

        ChannelFuture cf = b.connect().sync();

        cf.channel().closeFuture().sync();
    } finally {
        group.shutdownGracefully().sync();
    }
}

@Test
public void customEventLoopGroup_NotClosedWhenClientIsClosed() throws Exception {

    ThreadFactory threadFactory = spy(new CustomThreadFactory());
    // Cannot use DefaultEventLoopGroupFactory because the concrete
    // implementation it creates is platform-dependent and could be a final
    // (i.e. non-spyable) class.
    EventLoopGroup eventLoopGroup = spy(new NioEventLoopGroup(0, threadFactory));
    EventLoopGroupConfiguration eventLoopGroupConfiguration =
            EventLoopGroupConfiguration.builder()
                                       .eventLoopGroup(eventLoopGroup)
                                       .build();
    SdkAsyncHttpClient customClient =
            NettySdkHttpClientFactory.builder()
                                     .trustAllCertificates(true)
                                     .eventLoopGroupConfiguration(eventLoopGroupConfiguration)
                                     .build()
                                     .createHttpClient();

    makeSimpleRequest(customClient);
    customClient.close();

    Mockito.verify(threadFactory, atLeastOnce()).newThread(Mockito.any());
    Mockito.verify(eventLoopGroup, never()).shutdownGracefully();
}

public FabricClient(
    RpcConfig config,
    EventLoopGroup eventLoop,
    BufferAllocator allocator,
    FabricIdentity remoteIdentity,
    FabricIdentity localIdentity,
    FabricMessageHandler handler,
    FabricConnectionManager.CloseHandlerCreator closeHandlerFactory) {
  super(
      config,
      allocator.getAsByteBufAllocator(),
      eventLoop,
      RpcType.HANDSHAKE,
      FabricHandshake.class,
      FabricHandshake.PARSER);
  this.localIdentity = localIdentity;
  this.remoteIdentity = remoteIdentity;
  this.handler = handler;
  this.closeHandlerFactory = closeHandlerFactory;
  this.allocator = allocator;
}

/**
 * Adds a channel that listens locally
 */
public SocketAddress addLocalEndpoint()
{
    ChannelFuture channelfuture;

    synchronized (this.endpoints)
    {
        channelfuture = ((ServerBootstrap)((ServerBootstrap)(new ServerBootstrap()).channel(LocalServerChannel.class)).childHandler(new ChannelInitializer<Channel>()
        {
            protected void initChannel(Channel p_initChannel_1_) throws Exception
            {
                NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.SERVERBOUND);
                networkmanager.setNetHandler(new NetHandlerHandshakeMemory(NetworkSystem.this.mcServer, networkmanager));
                NetworkSystem.this.networkManagers.add(networkmanager);
                p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
            }
        }).group((EventLoopGroup)eventLoops.getValue()).localAddress(LocalAddress.ANY)).bind().syncUninterruptibly();
        this.endpoints.add(channelfuture);
    }

    return channelfuture.channel().localAddress();
}

/**
 * Attempts to determine the {@link Channel} class that corresponds to the given
 * event loop group.
 *
 * @param eventLoopGroup the event loop group to determine the {@link Channel} for
 * @return A {@link Channel} class for the given event loop group.
 */
public static Class<? extends Channel> resolveSocketChannelClass(EventLoopGroup eventLoopGroup) {
    if (eventLoopGroup instanceof DelegatingEventLoopGroup) {
        return resolveSocketChannelClass(((DelegatingEventLoopGroup) eventLoopGroup).getDelegate());
    }
    if (eventLoopGroup instanceof NioEventLoopGroup) {
        return NioSocketChannel.class;
    }
    if (eventLoopGroup instanceof EpollEventLoopGroup) {
        return EpollSocketChannel.class;
    }
    String socketFqcn = KNOWN_EL_GROUPS.get(eventLoopGroup.getClass().getName());
    if (socketFqcn == null) {
        throw new IllegalArgumentException("Unknown event loop group : " + eventLoopGroup.getClass());
    }
    return invokeSafely(() -> (Class<? extends Channel>) Class.forName(socketFqcn));
}

public void startUDPServer() throws InterruptedException {
    logger.info("Discovery UDPListener started");
    EventLoopGroup group = new NioEventLoopGroup(1);

    while (!shutdown) {
        Bootstrap bootstrap = this.createBootstrap(group);

        channel = bootstrap.bind(address, port).sync().channel();
        channel.closeFuture().sync();

        logger.warn("UDP channel closed. Recreating after 5 sec pause...");
        TimeUnit.SECONDS.sleep(5);
    }

    group.shutdownGracefully().sync();
}

@PostConstruct
public void start() {
    new Thread(() -> {
        logger.info("HttpProxyServer started on port: {}", port);
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(channelInitializer)
                    .bind(port).sync().channel().closeFuture().sync();
        } catch (InterruptedException e) {
            logger.error("shit happens", e);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }).start();
}

UserRPCServer(
    BootStrapContext context,
    Provider<SabotContext> dbContext,
    Provider<UserWorker> worker,
    BufferAllocator allocator,
    EventLoopGroup eventLoopGroup,
    InboundImpersonationManager impersonationManager
    ) {
  super(UserRpcConfig.getMapping(context.getConfig(), context.getExecutor()),
      allocator.getAsByteBufAllocator(),
      eventLoopGroup);
  this.worker = worker;
  this.dbContext = dbContext;
  this.allocator = allocator;
  this.impersonationManager = impersonationManager;
}

public static void main(String[] args) throws Exception {
    // Configure the client.
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(UkcpClientChannel.class)
                .handler(new ChannelInitializer<UkcpChannel>() {
                    @Override
                    public void initChannel(UkcpChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new KcpRttClientHandler(COUNT));
                    }
                });
        ChannelOptionHelper.nodelay(b, true, 20, 2, true)
                .option(UkcpChannelOption.UKCP_MTU, 512);

        // Start the client.
        ChannelFuture f = b.connect(HOST, PORT).sync();

        // Wait until the connection is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down the event loop to terminate all threads.
        group.shutdownGracefully();
    }
}

@Test
public void testTryWithResourcesShouldCloseAllClustersButNotEventLoopAndTimerIfProvided()
    throws Exception {
  EventLoopGroup eventLoop = new DefaultEventLoopGroup();
  Timer timer = new HashedWheelTimer();
  BoundCluster cluster;
  MockClient client;

  try (Server server =
      Server.builder()
          .withAddressResolver(localAddressResolver)
          .withTimer(timer)
          .withEventLoopGroup(eventLoop, LocalServerChannel.class)
          .build()) {

    cluster = server.register(ClusterSpec.builder().withNodes(5));
    BoundNode node = cluster.node(0);
    SocketAddress address = node.getAddress();
    client = new MockClient(eventLoop);
    client.connect(address);
  }

  // event loop should not have been closed.
  assertThat(eventLoop.isShutdown()).isFalse();
  // timer should not have since a custom one was not provided.
  cluster
      .getServer()
      .timer
      .newTimeout(
          timeout -> {
            // noop
          },
          1,
          TimeUnit.SECONDS);

  eventLoop.shutdownGracefully(0, 0, TimeUnit.SECONDS);
  timer.stop();
}

public static void main(String[] args) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                    .channel(NioDatagramChannel.class)
                    .option(ChannelOption.SO_BROADCAST, true)
                    .handler(new UdpClientHandler());

            Channel ch = b.bind(0).sync().channel();

            String UdpMsg = "this a message form client !";

            logger.info("Client ===> : {}", UdpMsg);

            // Broadcast the QOTM request to port 8880.
            ch.writeAndFlush(new DatagramPacket(
                    Unpooled.copiedBuffer(UdpMsg, CharsetUtil.UTF_8),
                    SocketUtils.socketAddress("localhost", PORT))).sync();

            // UdpClientHandler 接收到服务器响应的消息时，会关闭 DatagramChannel (数据报通道)
            // 如果 DatagramChannel (数据报通道) 没有在 5秒内关闭，那么打印错误日志，并退出
            if (!ch.closeFuture().await(5000)) {
                logger.error("Server response timed out.");
            }
        } finally {
            group.shutdownGracefully();
        }
    }

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

@Override
public void start() throws Exception {
  final EventLoopGroup eventLoopGroup = TransportCheck
      .createEventLoopGroup(context.getConfig().getInt(ExecConstants.USER_SERVER_RPC_THREADS), "UserServer-");

  this.eventLoopCloseable = new EventLoopCloseable(eventLoopGroup);
  this.allocator = context.getAllocator().newChildAllocator(
      "rpc:user",
      context.getConfig().getLong("dremio.exec.rpc.user.server.memory.reservation"),
      context.getConfig().getLong("dremio.exec.rpc.user.server.memory.maximum"));

  this.server = new UserRPCServer(context, dbContext, worker, allocator, eventLoopGroup, impersonationManager);

  Metrics.registerGauge("rpc.user.current", new Gauge<Long>() {
    @Override
    public Long getValue() {
      return allocator.getAllocatedMemory();
    }
  });
  Metrics.registerGauge("rpc.user.peak", new Gauge<Long>() {
    @Override
    public Long getValue() {
      return allocator.getPeakMemoryAllocation();
    }
  });
  int initialPort = context.getConfig().getInt(ExecConstants.INITIAL_USER_PORT);
  if(allowPortHunting){
    initialPort += 333;
  }

  port = server.bind(initialPort, allowPortHunting);
}

FastdfsPoolGroup(EventLoopGroup loopGroup, long connectTimeout, long readTimeout,
                 long idleTimeout, int maxConnPerHost) {
    this.loopGroup = loopGroup;
    this.connectTimeout = connectTimeout;
    this.readTimeout = readTimeout;
    this.idleTimeout = idleTimeout;
    this.maxConnPerHost = maxConnPerHost;
}

/**
 * Create a new NetworkManager from the server host and connect it to the server
 */
public static NetworkManager createNetworkManagerAndConnect(InetAddress address, int serverPort, boolean useNativeTransport)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    Class <? extends SocketChannel > oclass;
    LazyLoadBase <? extends EventLoopGroup > lazyloadbase;

    if (Epoll.isAvailable() && useNativeTransport)
    {
        oclass = EpollSocketChannel.class;
        lazyloadbase = CLIENT_EPOLL_EVENTLOOP;
    }
    else
    {
        oclass = NioSocketChannel.class;
        lazyloadbase = CLIENT_NIO_EVENTLOOP;
    }

    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)lazyloadbase.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            try
            {
                p_initChannel_1_.config().setOption(ChannelOption.TCP_NODELAY, Boolean.valueOf(true));
            }
            catch (ChannelException var3)
            {
                ;
            }

            p_initChannel_1_.pipeline().addLast((String)"timeout", (ChannelHandler)(new ReadTimeoutHandler(30))).addLast((String)"splitter", (ChannelHandler)(new NettyVarint21FrameDecoder())).addLast((String)"decoder", (ChannelHandler)(new NettyPacketDecoder(EnumPacketDirection.CLIENTBOUND))).addLast((String)"prepender", (ChannelHandler)(new NettyVarint21FrameEncoder())).addLast((String)"encoder", (ChannelHandler)(new NettyPacketEncoder(EnumPacketDirection.SERVERBOUND))).addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(oclass)).connect(address, serverPort).syncUninterruptibly();
    return networkmanager;
}

public void start() throws Exception {
    // Configure SSL.
    final SslContext sslCtx;
    if (SSL) {
        SelfSignedCertificate ssc = new SelfSignedCertificate();
        sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
    } else {
        sslCtx = null;
    }

    // Configure the server.
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100)
                .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new FileServerHandlerInitializer());

        // Start the server.
        ChannelFuture f = b.bind(getHostAddress(), PORT).sync();
        // System.out.println("server is started "+f.isSuccess());
        setStarted(true);
        // Wait until the server socket is closed.
        f.channel().closeFuture().sync();
    } finally {
        // Shut down all event loops to terminate all threads.
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

}

@Test
public void referenceCountIsOnceDecrementedOnClose() {
    EventLoopGroup group = SharedEventLoopGroup.get();
    group.shutdownGracefully();
    assertThat(SharedEventLoopGroup.referenceCount()).isEqualTo(0);
    group.shutdownGracefully();
    assertThat(SharedEventLoopGroup.referenceCount()).isEqualTo(0);
}

public EventLoopGroup getChildEventLoopGroup(){
    EventLoopGroup child = childGroupMap.get(protocolType.value()) ;
    if(child == null){
        child = initChildEventLoopGroup();
        childGroupMap.put(protocolType.value(),child);
    }
    return child;
}

public static void start(MemberEventLoop loop) throws InterruptedException {
    String host = "127.0.0.1";
    int port = 9005;

    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(workerGroup);
        b.channel(NioSocketChannel.class);
        b.option(ChannelOption.SO_KEEPALIVE, true);
        b.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {

                ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                ch.pipeline().addLast(new ProtobufDecoder(SocketMessage.getDefaultInstance()));

                ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                ch.pipeline().addLast(new ProtobufEncoder());

                ch.pipeline().addLast(new IdleStateHandler(0, 5, 10, TimeUnit.SECONDS));
                ch.pipeline().addLast(new BusinessRouterHandler(loop));
            }
        });

        // Start the client.
        ChannelFuture f = b.connect(host, port).sync();

        // Wait until the connection is closed.

        f.channel().closeFuture().sync();

    } finally {
        workerGroup.shutdownGracefully();
    }

}

public void addPublicEndpoint(InetAddress address, int port)
{
    synchronized (this.endpoints)
    {
        final EventLoopGroup boss = new NioEventLoopGroup();
        final EventLoopGroup worker = new NioEventLoopGroup();
        final ServerBootstrap b = new ServerBootstrap()
                .group(boss, worker)
                .childHandler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception
                    {
                        final PacketRegistry registry = NetworkEngine.this.packetRegistry;
                        final NetworkDispatcher dispatch = new NetworkDispatcher(
                                NetworkEngine.this, NetworkSide.SERVER);
                        NetworkEngine.this.dispatchers.add(dispatch);

                        ch.pipeline()
                                .addLast(new VarInt21FrameDecoder())
                                .addLast(new PacketDecoder(NetworkSide.SERVER, registry))
                                .addLast(new VarInt21FrameEncoder())
                                .addLast(new PacketEncoder(NetworkSide.CLIENT, registry))
                                .addLast(dispatch);
                    }
                })
                .channel(NioServerSocketChannel.class);

        //Host and wait until done
        b.localAddress(address, port).bind().syncUninterruptibly();
    }
}

public static void main(String[] args) throws Exception {
    // Configure SSL.
    //final SslContext sslCtx;
    ConfigurationManager.getConfigInstance().setProperty("app.localLog.path","E:\\Tools\\apache-tomcat-7.0.68\\logs");

    // Configure the server.
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.option(ChannelOption.SO_BACKLOG, 1024);
        b.group(bossGroup, workerGroup)
         .channel(NioServerSocketChannel.class)
         .handler(new LoggingHandler(LogLevel.INFO))
         .childHandler(new HttpHelloWorldServerInitializer());


        Channel ch = b.bind(PORT).sync().channel();



        System.err.println("Open your web browser and navigate to " +
                (SSL? "https" : "http") + "://127.0.0.1:" + PORT + '/');

        ch.closeFuture().sync();

    } finally {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

private synchronized static Pair<EventLoopGroup, Class<? extends Channel>>
    getGlobalEventLoopGroup(Configuration conf) {
  if (GLOBAL_EVENT_LOOP_GROUP == null) {
    GLOBAL_EVENT_LOOP_GROUP = createEventLoopGroup(conf);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Create global event loop group "
          + GLOBAL_EVENT_LOOP_GROUP.getFirst().getClass().getSimpleName());
    }
  }
  return GLOBAL_EVENT_LOOP_GROUP;
}

/**
 * Prepares a clientside NetworkManager: establishes a connection to the socket supplied and configures the channel
 * pipeline. Returns the newly created instance.
 */
public static NetworkManager provideLocalClient(SocketAddress address)
{
    final NetworkManager networkmanager = new NetworkManager(EnumPacketDirection.CLIENTBOUND);
    ((Bootstrap)((Bootstrap)((Bootstrap)(new Bootstrap()).group((EventLoopGroup)CLIENT_LOCAL_EVENTLOOP.getValue())).handler(new ChannelInitializer<Channel>()
    {
        protected void initChannel(Channel p_initChannel_1_) throws Exception
        {
            p_initChannel_1_.pipeline().addLast((String)"packet_handler", (ChannelHandler)networkmanager);
        }
    })).channel(LocalChannel.class)).connect(address).syncUninterruptibly();
    return networkmanager;
}

public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup();
        final Bootstrap cb = new Bootstrap();
        cb.group(group).channel(NioSocketChannel.class);
        InetSocketAddress addr1 = new InetSocketAddress("10.0.0.10", 8888);
        InetSocketAddress addr2 = new InetSocketAddress("10.0.0.11", 8888);

        //连接池map
        ChannelPoolMap<InetSocketAddress, SimpleChannelPool> poolMap = new AbstractChannelPoolMap<InetSocketAddress, SimpleChannelPool>() {
            @Override
            protected SimpleChannelPool newPool(InetSocketAddress key) {
                return new SimpleChannelPool(cb.remoteAddress(key), new TestChannelPoolHandler());
            }
        };

        final SimpleChannelPool pool1 = poolMap.get(addr1);//取出連接addr1地址的连接池
        final SimpleChannelPool pool2 = poolMap.get(addr2);//取出連接addr2地址的连接池
        Future<Channel> f1 = pool1.acquire();//获取一个连接
        f1.addListener(new FutureListener<Channel>() {
            @Override
            public void operationComplete(Future<Channel> f) {
                if (f.isSuccess()) {
                    Channel ch = f.getNow();
                   //连接地址1的某个channel
                    //使用连接发送消息
//                  ch.write(msg)
                    //用完释放
                    pool1.release(ch);
                }
            }
        });

    }

/**
 * Starts the client that means starting the {@link NioEventLoopGroup}
 * and connecting the {@link #bootstrap}
 *
 * @throws Exception If something goes wrong
 */
public void connect() throws Exception {
    EventLoopGroup eventExecutors = new NioEventLoopGroup();

    try {
        this.channel = bootstrap.connect(getHost(), getPort()).sync().channel();

        channel.closeFuture().sync().syncUninterruptibly();
    }
    finally {
        eventExecutors.shutdownGracefully();
    }
}

@Override
public EventLoopGroup create() {
    int numThreads = numberOfThreads == null ? 0 : numberOfThreads;
    return new NioEventLoopGroup(numThreads, resolveThreadFactory());
    /*
    Need to investigate why epoll is raising channel inactive after succesful response that causes
    problems with retries.

    if (Epoll.isAvailable() && isNotAwsLambda()) {
        return new EpollEventLoopGroup(numThreads, resolveThreadFactory());
    } else {

    }*/
}

public UserClient(DrillConfig config, boolean supportComplexTypes, BufferAllocator alloc,
    EventLoopGroup eventLoopGroup) {
  super(
      UserRpcConfig.getMapping(config),
      alloc,
      eventLoopGroup,
      RpcType.HANDSHAKE,
      BitToUserHandshake.class,
      BitToUserHandshake.PARSER,
      "user client");
  this.supportComplexTypes = supportComplexTypes;
}

private EventLoopGroup newEventLoop(int size, String threadNamePrefix) {
    switch (transport) {
        case EPOLL: {
            return new EpollEventLoopGroup(size, new HekateThreadFactory(threadNamePrefix));
        }
        case NIO: {
            return new NioEventLoopGroup(size, new HekateThreadFactory(threadNamePrefix));
        }
        case AUTO: // <-- Fail since AUTO must be resolved in the constructor.
        default: {
            throw new IllegalArgumentException("Unexpected transport type: " + transport);
        }
    }
}

public static EventLoopGroup createEventLoopGroup(int nThreads, String prefix) {
   if(SUPPORTS_EPOLL){
     return new EpollEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }else{
     return new NioEventLoopGroup(nThreads, new NamedThreadFactory(prefix));
   }
}

protected EventLoopGroup newEventLoop(int thread) {
    if (context().transport() == NetworkTransportType.EPOLL) {
        return new EpollEventLoopGroup(thread);
    } else {
        return new NioEventLoopGroup(thread);
    }
}

@SuppressWarnings({ "rawtypes", "unchecked" })
@Before
public void setUp() throws Exception {
    factory = new NettyThreadgroupModuleFactory();
    super.initConfigTransactionManagerImpl(new HardcodedModuleFactoriesResolver(mockedContext,factory));

    Filter mockFilter = mock(Filter.class);
    doReturn("mock").when(mockFilter).toString();
    doReturn(mockFilter).when(mockedContext).createFilter(anyString());
    doNothing().when(mockedContext).addServiceListener(any(ServiceListener.class), anyString());
    ServiceReference mockServiceRef = mock(ServiceReference.class);
    doReturn(new ServiceReference[]{mockServiceRef}).when(mockedContext).
            getServiceReferences(anyString(), anyString());
    doReturn(mock(EventLoopGroup.class)).when(mockedContext).getService(mockServiceRef);
}