Java 类io.netty.channel.socket.nio.NioSocketChannel 实例源码

public void start() throws InterruptedException {
    final EventLoopGroup workerGroup = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();

    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(workerGroup)
                .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                .handler(new OpenCloudChannelInitializer(this))
                .connect(this.host, this.port).sync().channel().closeFuture().syncUninterruptibly();
    } catch (Exception ex) {
        if (ex.getClass().getSimpleName().equals("AnnotatedConnectException")) {
            System.err.println("Cannot connect to master!");
            channel.close();
        } else {
            ex.printStackTrace();
        }
    } finally {
        workerGroup.shutdownGracefully();
        System.out.println("Netty client stopped");
        Runtime.getRuntime().halt(0);
    }
}

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();
    }
}

@Override
    public void open() {
        EventLoopGroup eventLoop = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(eventLoop);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3 * 1000);
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            public void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline()
//                        .addLast("logging",new LoggingHandler(LogLevel.INFO))
                        .addLast("decoder", new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader()))) // in 1
                        .addLast("handler", new ClientReadHandler()) // in 2
                        .addLast("encoder", new ObjectEncoder())// out 3
                        .addLast("idleStateHandler", new IdleStateHandler(0, 1, 0))
                        .addLast(new ClientIdleHandler());

            }
        });
    }

@Override
protected FastdfsPool newPool(InetSocketAddress addr) {
    if (LOG.isDebugEnabled()) {
        LOG.debug("channel pool created : {}", addr);
    }

    Bootstrap bootstrap = new Bootstrap().channel(NioSocketChannel.class).group(loopGroup);
    bootstrap.remoteAddress(addr);
    bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, (int) connectTimeout);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);
    bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
    return new FastdfsPool(
            bootstrap,
            readTimeout,
            idleTimeout,
            maxConnPerHost
    );
}

public void start() {
    bootstrap.group(group).channel(NioSocketChannel.class);
    bootstrap.option(ChannelOption.TCP_NODELAY, true);
    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            //                ch.pipeline().addLast(new IdleStateHandler(1, 1, 5));
            ch.pipeline().addLast(new KyroMsgDecoder());
            ch.pipeline().addLast(new KyroMsgEncoder());
            ch.pipeline().addLast(new ClientHandler());
        }
    });

    new ScheduledThreadPoolExecutor(1).scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            scanResponseTable(3000);
        }
    }, 1000, 1000, TimeUnit.MILLISECONDS);
}

public static void main(String[] args) throws IOException, InterruptedException {
    Bootstrap b = new Bootstrap();
    b.group(new NioEventLoopGroup())
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<NioSocketChannel>() {
                @Override
                protected void initChannel(NioSocketChannel ch) throws Exception {
                }
            });
    b.connect("localhost", 8090).addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            if (future.isSuccess()) {
                future.channel().write(Unpooled.buffer().writeBytes("123".getBytes()));
                future.channel().flush();
                future.channel().close();
            }
        }
    });
}

private BinaryClient(BinaryClientBuilder builder) throws Exception {
    this.clientName = builder.clientName;
    this.remoteAddress = Objects.requireNonNull(builder.remoteAddress, "remoteAddress");
    this.autoReconnect = builder.autoReconnect;
    this.decoder = Objects.requireNonNull(builder.decoder, "decoder");
    this.encoder = Objects.requireNonNull(builder.encoder, "encoder");
    this.factory = Objects.requireNonNull(builder.factory, "factory");
    this.onChannelStateChanged = builder.onChannelStateChanged;
    this.onExceptionCaught = builder.onExceptionCaught;
    this.onConnectionEffective = builder.onConnectionEffective;
    this.dispatchMessage = builder.dispatchMessage;
    this.heartIntervalSec = builder.heartIntervalSec;
    // 内部消息注册
    factory.registerMsg(new ConnectionValidateServerHandler())
            .registerMsg(new ConnectionValidateSuccessServerHandler()).registerMsg(new HeartServerHandler());
    decodeUtil = SymmetricEncryptionUtil.getDecodeInstance(remoteAddress.getPass());
    bootstrap = new Bootstrap();
    bootstrap.channel(NioSocketChannel.class);
    log.info(clientName + " nio init");
    bootstrap.group(group).option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .handler(new ChannelInitializerImpl());
}

@Override
public void connect() throws IOException, InterruptedException {
    workerGroup = new NioEventLoopGroup();

    Bootstrap b = new Bootstrap();
    b.group(workerGroup);
    b.channel(NioSocketChannel.class);
    b.option(ChannelOption.SO_KEEPALIVE, true);
    b.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline p = ch.pipeline();
            p.addLast(
                    //new LoggingHandler(LogLevel.INFO),
                    new MsgEncoder(),
                    new MsgDecoder(),
                    new NettyClientHandler()
            );
        }
    });

    ChannelFuture f = b.connect(address, port).sync();
    channel = f.channel();
}

private final Future<Void> bootstrap(final NettyChannel channel) {
  final Promise<Void> p = Promise.apply();
  new Bootstrap().group(eventLoopGroup).channel(NioSocketChannel.class)
      .option(ChannelOption.SO_KEEPALIVE, true)
      .option(ChannelOption.AUTO_READ, false)
      .handler(new ChannelInitializer<io.netty.channel.Channel>() {
        @Override
        protected void initChannel(final io.netty.channel.Channel ch) throws Exception {
          ch.pipeline().addLast(new MessageDecoder(), new MessageEncoder(),
              new FlowControlHandler(), channel);
        }
      })
      .connect(new InetSocketAddress(host, port))
      .addListener(future -> p.become(Future.VOID));
  return p;
}

public ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
    ethereumListener.trace("Connecting to: " + host + ":" + port);

    EthereumChannelInitializer ethereumChannelInitializer = ctx.getBean(EthereumChannelInitializer.class, remoteId);
    ethereumChannelInitializer.setPeerDiscoveryMode(discoveryMode);

    Bootstrap b = new Bootstrap();
    b.group(workerGroup);
    b.channel(NioSocketChannel.class);

    b.option(ChannelOption.SO_KEEPALIVE, true);
    b.option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT);
    b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.peerConnectionTimeout());
    b.remoteAddress(host, port);

    b.handler(ethereumChannelInitializer);

    // Start the client.
    return b.connect();
}

@Override
public Future<Channel> getConnection(HostAndPort address)
{
    try {
        Bootstrap bootstrap = new Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .option(CONNECT_TIMEOUT_MILLIS, saturatedCast(connectTimeout.toMillis()))
                .handler(new ThriftClientInitializer(
                        messageFraming,
                        messageEncoding,
                        requestTimeout,
                        socksProxy,
                        sslContextSupplier));

        Promise<Channel> promise = group.next().newPromise();
        bootstrap.connect(new InetSocketAddress(address.getHost(), address.getPort()))
                .addListener((ChannelFutureListener) future -> notifyConnect(future, promise));
        return promise;
    }
    catch (Throwable e) {
        return group.next().newFailedFuture(new TTransportException(e));
    }
}

/**
 * 初始化连接池
 */
public void init() {
    bootstrap = new Bootstrap();
    eventLoopGroup = new NioEventLoopGroup();
    bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(new LoggingHandler());
    //所有的公用一个eventloopgroup, 对于客户端来说应该问题不大!
    poolMap = new AbstractChannelPoolMap<InetSocketAddress, FixedChannelPool>() {
        @Override
        protected FixedChannelPool newPool(InetSocketAddress key) {
            return new FixedChannelPool(bootstrap.remoteAddress(key), new FixedChannelPoolHandler(), 2);
        }
    };
    //预先建立好链接
    serverListConfig.getAddressList().stream().forEach(address -> {
        poolMap.get(address);
    });
}

public void run() throws Exception {
    NioEventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(host, port))
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoClientHandler());
                    }
                });
        ChannelFuture f = b.connect().sync();
        f.channel().closeFuture().sync();

    } finally {
        group.shutdownGracefully().sync();
    }
}

public ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
    ethereumListener.trace("Connecting to: " + host + ":" + port);

    EthereumChannelInitializer ethereumChannelInitializer = ethereumChannelInitializerFactory.newInstance(remoteId);
    ethereumChannelInitializer.setPeerDiscoveryMode(discoveryMode);

    Bootstrap b = new Bootstrap();
    b.group(workerGroup);
    b.channel(NioSocketChannel.class);

    b.option(ChannelOption.SO_KEEPALIVE, true);
    b.option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT);
    b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.peerConnectionTimeout());
    b.remoteAddress(host, port);

    b.handler(ethereumChannelInitializer);

    // Start the client.
    return b.connect();
}

public void start(String hostName, int port) {
    Executors.newSingleThreadExecutor().submit(() -> {
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new KyChannelInitializer());
        if (hostName != null && !hostName.equals(""))
            bootstrap.remoteAddress(new InetSocketAddress(hostName, port));
        else
            bootstrap.remoteAddress(new InetSocketAddress(port));
        ChannelFuture channelFuture = null;

        try {
            channelFuture = bootstrap.connect().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        startListenerHandle(channelFuture, launchListener);
    });
}

public void start() {
    new Thread(() -> {
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .remoteAddress(new InetSocketAddress(30232)) //"woodswang",
                .handler(KyChannelInitializer.newInstance());
        ChannelFuture channelFuture = null;

        try {
            channelFuture = bootstrap.connect().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        startListenerHandle(channelFuture, launchListener);
    }).start();
}

private void startClient(String hostName, int port, int id) {
    Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(group)
            .attr(channelId, id)
            .channel(NioSocketChannel.class)
            .handler(clientChannelInitializer);
    if (hostName != null && !"".equals(hostName)) {
        bootstrap.remoteAddress(new InetSocketAddress(hostName, port));
    } else {
        bootstrap.remoteAddress(new InetSocketAddress(port));
    }
    try {
        bootstrap.connect().sync().addListener(future -> logger.info("Channel「" + id + "」" + "已连接"));
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

public EchoClient(String host, int port) {
    EventLoopGroup worker = new NioEventLoopGroup();

    Bootstrap b = new Bootstrap();

    b.group(worker)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) {
                    socketChannel.pipeline()
                            .addLast(new StringDecoder())
                            .addLast(new StringEncoder())
                            .addLast(ech);
                }
            });

    b.connect(host, port);
}

@Override
public void startUp(FloodlightModuleContext context) 
        throws FloodlightModuleException {
    shutdown = false;
    workerExecutor = new NioEventLoopGroup();
    timer = new HashedWheelTimer();

    pipelineFactory = new RemoteSyncChannelInitializer(timer, this);

    final Bootstrap bootstrap = new Bootstrap()
    .channel(NioSocketChannel.class)
    .group(workerExecutor)
    .option(ChannelOption.SO_REUSEADDR, true)
    .option(ChannelOption.SO_KEEPALIVE, true)
    .option(ChannelOption.TCP_NODELAY, true)
    .option(ChannelOption.SO_SNDBUF, RPCService.SEND_BUFFER_SIZE)
    .option(ChannelOption.SO_RCVBUF, RPCService.SEND_BUFFER_SIZE)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, RPCService.CONNECT_TIMEOUT)
    .handler(pipelineFactory);

    clientBootstrap = bootstrap;
}

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() throws Exception {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
         .channel(NioSocketChannel.class)
         .remoteAddress(new InetSocketAddress(host, port))
         .handler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) 
                 throws Exception {
                 ch.pipeline().addLast(
                         new EchoClientHandler());
             }
         });

        ChannelFuture f = bootstrap.connect().sync();

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

@Override
public boolean start() {
    if (nettyChannel != null)
        return false;

    group = new NioEventLoopGroup();
    try {
        final Bootstrap bootstrap = new io.netty.bootstrap.Bootstrap()
                .group(group)
                .channel(NioSocketChannel.class)
                .handler(new NettyClientInitializer(this::trigger));

        nettyChannel = bootstrap.connect(host, port).sync().channel();
    } catch (final Exception e) {
        LOGGER.error("Impossible to connect to {}:{}", host, port);
        return false;
    }
    LOGGER.info("Connected on {}:{}", host, port);

    return true;
}

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();
    }
}

public static void main(String[] args) throws Exception {
    // Configure the client.
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline p = ch.pipeline();
                        p.addLast(new TcpRttDecoder())
                                .addLast(new TcpRttClientHandler(COUNT));
                    }
                }).option(ChannelOption.TCP_NODELAY, true);

        // 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();
    }
}

/**
 * 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 init() throws SyncException {
    cg = new DefaultChannelGroup("Cluster Bootstrap", GlobalEventExecutor.INSTANCE);

    workerExecutor = new NioEventLoopGroup();
    timer = new HashedWheelTimer();

    bootstrap = new Bootstrap()
    .group(workerExecutor)
    .channel(NioSocketChannel.class)
    .option(ChannelOption.SO_REUSEADDR, true)
    .option(ChannelOption.SO_KEEPALIVE, true)
    .option(ChannelOption.TCP_NODELAY, true)
    .option(ChannelOption.SO_SNDBUF, RPCService.SEND_BUFFER_SIZE)
    .option(ChannelOption.SO_RCVBUF, RPCService.SEND_BUFFER_SIZE)
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, RPCService.CONNECT_TIMEOUT);

    pipelineFactory = new BootstrapChannelInitializer(timer, this);
    bootstrap.handler(pipelineFactory);
}

/**
 * Init Bootstrap
 */
public static final Bootstrap getBootstrap() {
    EventLoopGroup group = new NioEventLoopGroup();
    Bootstrap b = new Bootstrap();
    b.group(group);
    b.channel(NioSocketChannel.class);
    b.handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast("frameDecoder", new ProtobufVarint32FrameDecoder());
            pipeline.addLast("decoder", new ProtobufDecoder(MessageBuf.JMTransfer.getDefaultInstance()));
            pipeline.addLast("frameEncoder", new ProtobufVarint32LengthFieldPrepender());
            pipeline.addLast("encoder", new ProtobufEncoder());
            pipeline.addLast("handler", new TcpClientHandler());
        }
    });


    b.option(ChannelOption.SO_KEEPALIVE, true);
    return b;
}

public ChannelFuture connectAsync(String host, int port, String remoteId, boolean discoveryMode) {
    ethereumListener.trace("Connecting to: " + host + ":" + port);

    EthereumChannelInitializer ethereumChannelInitializer = ctx.getBean(EthereumChannelInitializer.class, remoteId);
    ethereumChannelInitializer.setPeerDiscoveryMode(discoveryMode);

    Bootstrap b = new Bootstrap();
    b.group(workerGroup);
    b.channel(NioSocketChannel.class);

    b.option(ChannelOption.SO_KEEPALIVE, true);
    b.option(ChannelOption.MESSAGE_SIZE_ESTIMATOR, DefaultMessageSizeEstimator.DEFAULT);
    b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, config.peerConnectionTimeout());
    b.remoteAddress(host, port);

    b.handler(ethereumChannelInitializer);

    // Start the client.
    return b.connect();
}

@Override
public void connect(final InetSocketAddress socketAddress) {
    workerGroup = new NioEventLoopGroup(workerGroupThreads);
    Bootstrap bootstrap = new Bootstrap();
    try {
        bootstrap
            .group(workerGroup)
            .channel(NioSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.TCP_NODELAY, true)
            .handler(clientChannelInitializer);
    } catch (final Exception ex) {
        throw new ClientException(ex);
    }
    channel = bootstrap.connect(socketAddress.getAddress().getHostAddress(), socketAddress.getPort()).syncUninterruptibly().channel();
}

@Override
public void connect() {
  checkState(channel == null, "channel already initialized");
  try {
    TrustManagerFactory trustFactory = TrustManagerFactory.getInstance(
        TrustManagerFactory.getDefaultAlgorithm());
    trustFactory.init((KeyStore) null);
    final SslContext sslContext = SslContextBuilder.forClient()
        .trustManager(trustFactory).build();
    Bootstrap bootstrap = new Bootstrap();
    final int port = uri.getPort() != -1 ? uri.getPort() : DEFAULT_WSS_PORT;
    bootstrap.group(group)
        .channel(NioSocketChannel.class)
        .handler(new ChannelInitializer<SocketChannel>() {
          @Override
          protected void initChannel(SocketChannel ch) {
            ChannelPipeline p = ch.pipeline();
            p.addLast(sslContext.newHandler(ch.alloc(), uri.getHost(), port));
            p.addLast(
                new HttpClientCodec(),
                // Set the max size for the HTTP responses. This only applies to the WebSocket
                // handshake response from the server.
                new HttpObjectAggregator(32 * 1024),
                channelHandler);
          }
        });

    ChannelFuture channelFuture = bootstrap.connect(uri.getHost(), port);
    this.channel = channelFuture.channel();
    channelFuture.addListener(
        new ChannelFutureListener() {
          @Override
          public void operationComplete(ChannelFuture future) throws Exception {
            if (!future.isSuccess()) {
              eventHandler.onError(future.cause());
            }
          }
        }
    );
  } catch (Exception e) {
    eventHandler.onError(e);
  }
}

/**
 * Create a new NetworkManager from the server host and connect it to the server
 */
@SideOnly(Side.CLIENT)
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;
}

@Override public void prepare(final Benchmark benchmark) {
  this.concurrencyLevel = benchmark.concurrencyLevel;
  this.targetBacklog = benchmark.targetBacklog;

  ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
    @Override public void initChannel(SocketChannel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();

      if (benchmark.tls) {
        SslClient sslClient = SslClient.localhost();
        SSLEngine engine = sslClient.sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
      }

      pipeline.addLast("codec", new HttpClientCodec());
      pipeline.addLast("inflater", new HttpContentDecompressor());
      pipeline.addLast("handler", new HttpChannel(channel));
    }
  };

  bootstrap = new Bootstrap();
  bootstrap.group(new NioEventLoopGroup(concurrencyLevel))
      .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .channel(NioSocketChannel.class)
      .handler(channelInitializer);
}

@Override public void prepare(final Benchmark benchmark) {
  this.concurrencyLevel = benchmark.concurrencyLevel;
  this.targetBacklog = benchmark.targetBacklog;

  ChannelInitializer<SocketChannel> channelInitializer = new ChannelInitializer<SocketChannel>() {
    @Override public void initChannel(SocketChannel channel) throws Exception {
      ChannelPipeline pipeline = channel.pipeline();

      if (benchmark.tls) {
        SslClient sslClient = SslClient.localhost();
        SSLEngine engine = sslClient.sslContext.createSSLEngine();
        engine.setUseClientMode(true);
        pipeline.addLast("ssl", new SslHandler(engine));
      }

      pipeline.addLast("codec", new HttpClientCodec());
      pipeline.addLast("inflater", new HttpContentDecompressor());
      pipeline.addLast("handler", new HttpChannel(channel));
    }
  };

  bootstrap = new Bootstrap();
  bootstrap.group(new NioEventLoopGroup(concurrencyLevel))
      .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
      .channel(NioSocketChannel.class)
      .handler(channelInitializer);
}

protected void initChannel(NioSocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    // 编解码 http 请求
    pipeline.addLast(new HttpServerCodec());
    // 写文件内容
    pipeline.addLast(new ChunkedWriteHandler());
    // 聚合解码 HttpRequest/HttpContent/LastHttpContent 到 FullHttpRequest
    // 保证接收的 Http 请求的完整性
    pipeline.addLast(new HttpObjectAggregator(64 * 1024));
    // 处理其他的 WebSocketFrame
    pipeline.addLast(new WebSocketServerProtocolHandler("/chat"));
    // 处理 TextWebSocketFrame
    pipeline.addLast(protoCodec);
    pipeline.addLast(serverHandler);
}

public void start() throws Exception {
    try {
        logger.info("[info] >>> start netty client.");
        bootstrap.group(workerGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(initializer);
        logger.info("[info] >>> set netty client params");
        bootstrap.option(ChannelOption.SO_BACKLOG, 65535);
        //keep connect
        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
        bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
        bootstrap.option(ChannelOption.TCP_NODELAY, true);


        channel = bootstrap.connect(host, port).syncUninterruptibly().channel();
    } catch (Exception e) {
        e.printStackTrace();
        logger.error("[info] >>> netty client start fail.");
    }
}

/**
 * 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;
}

private Bootstrap createBootstrap() {
    final Bootstrap bootstrap = new Bootstrap();
    bootstrap.group(new NioEventLoopGroup(workerCount, daemonThreadFactory(settings, TRANSPORT_CLIENT_BOSS_THREAD_NAME_PREFIX)));
    bootstrap.channel(NioSocketChannel.class);

    bootstrap.handler(getClientChannelInitializer());

    bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.toIntExact(defaultConnectionProfile.getConnectTimeout().millis()));
    bootstrap.option(ChannelOption.TCP_NODELAY, TCP_NO_DELAY.get(settings));
    bootstrap.option(ChannelOption.SO_KEEPALIVE, TCP_KEEP_ALIVE.get(settings));

    final ByteSizeValue tcpSendBufferSize = TCP_SEND_BUFFER_SIZE.get(settings);
    if (tcpSendBufferSize.getBytes() > 0) {
        bootstrap.option(ChannelOption.SO_SNDBUF, Math.toIntExact(tcpSendBufferSize.getBytes()));
    }

    final ByteSizeValue tcpReceiveBufferSize = TCP_RECEIVE_BUFFER_SIZE.get(settings);
    if (tcpReceiveBufferSize.getBytes() > 0) {
        bootstrap.option(ChannelOption.SO_RCVBUF, Math.toIntExact(tcpReceiveBufferSize.getBytes()));
    }

    bootstrap.option(ChannelOption.RCVBUF_ALLOCATOR, recvByteBufAllocator);

    final boolean reuseAddress = TCP_REUSE_ADDRESS.get(settings);
    bootstrap.option(ChannelOption.SO_REUSEADDR, reuseAddress);

    bootstrap.validate();

    return bootstrap;
}

public static Class<? extends SocketChannel> getClientSocketChannel(){
  if(SUPPORTS_EPOLL){
    return EpollSocketChannel.class;
  }else{
    return NioSocketChannel.class;
  }
}

@Override
public void run() {
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {

                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
                        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                        pipeline.addLast("decoder", new StringDecoder(CharsetUtil.UTF_8));
                        pipeline.addLast("encoder", new StringEncoder(CharsetUtil.UTF_8));
                        pipeline.addLast(new SimpleClientChannelHandler());
                    }

                });
        ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
        if (channelFuture.isSuccess()) {
            System.out.println(String.format("connect server(%s:%s) sucess", host, port));
        }
        channelFuture.channel().closeFuture().sync();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        group.shutdownGracefully();
    }
}

@Override
public void initChannel(NioSocketChannel ch) throws Exception {
    try {
        if (!peerDiscoveryMode) {
            logger.debug("Open {} connection, channel: {}", isInbound() ? "inbound" : "outbound", ch.toString());
        }

        if (isInbound() && channelManager.isRecentlyDisconnected(ch.remoteAddress().getAddress())) {
            // avoid too frequent connection attempts
            logger.debug("Drop connection - the same IP was disconnected recently, channel: {}", ch.toString());
            ch.disconnect();
            return;
        }

        final org.talchain.net.server.Channel channel = ctx.getBean(org.talchain.net.server.Channel.class);
        channel.init(ch.pipeline(), remoteId, peerDiscoveryMode, channelManager);

        if(!peerDiscoveryMode) {
            channelManager.add(channel);
        }

        // limit the size of receiving buffer to 1024
        ch.config().setRecvByteBufAllocator(new FixedRecvByteBufAllocator(256 * 1024));
        ch.config().setOption(ChannelOption.SO_RCVBUF, 256 * 1024);
        ch.config().setOption(ChannelOption.SO_BACKLOG, 1024);

        // be aware of channel closing
        ch.closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                if (!peerDiscoveryMode) {
                    channelManager.notifyDisconnect(channel);
                }
            }
        });

    } catch (Exception e) {
        logger.error("Unexpected error: ", e);
    }
}