Java 类io.netty.channel.socket.SocketChannel 实例源码

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
protected AsyncRpcClient createRpcClientRTEDuringConnectionSetup(Configuration conf) {
  setConf(conf);
  return new AsyncRpcClient(conf, new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
          ch.pipeline().addFirst(new ChannelOutboundHandlerAdapter() {
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
                throws Exception {
              promise.setFailure(new RuntimeException("Injected fault"));
            }
          });
        }
      });
}

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

@Override
public void openServer(URL url) throws Exception{
    EventLoopGroup eventLoop = new NioEventLoopGroup();
    EventLoopGroup workLoop = new NioEventLoopGroup();
    serverBootstrap = new ServerBootstrap();
    serverBootstrap.group(eventLoop, workLoop);
    serverBootstrap.channel(NioServerSocketChannel.class);
    serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);
    serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>(){

        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline()
                    .addLast("decoder", new ObjectDecoder(ClassResolvers.cacheDisabled(getClass().getClassLoader()))) // in 1
                    .addLast("handler", new ServerHandler()) // in 2
                    .addLast("encoder", new ObjectEncoder()); // out 3

        }
    });
    serverChannel = serverBootstrap.bind(url.getPort()).sync().sync().channel();
    logger.info("start server at:" + url.getPort());
}

@Override
public void bind(int port) {
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);  
    EventLoopGroup workerGroup = new NioEventLoopGroup();  
    ServerBootstrap bootstrap = new ServerBootstrap()
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class).localAddress(new InetSocketAddress(8888))
            .childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new Encoder(serializer), new Decoder(serializer), new ProviderHandler());
                }

            });

    bootstrap.bind(port);
}

@Override
public void init() {
    super.init();
    b.group(bossGroup, workGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_BACKLOG, 1024)
            .localAddress(new InetSocketAddress(port))
            .childHandler(new ChannelInitializer<SocketChannel>() {

                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(defLoopGroup,
                            new SdkServerDecoder(12),  // 自定义解码器
                            new SdkServerEncoder(),    // 自定义编码器
                            new SdkServerHandler(snowFlake) // 自定义处理器
                    );
                }
            });
}

public void doOpen() throws InterruptedException {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try{
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup);
        serverBootstrap.channel(NioServerSocketChannel.class);
        serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
            protected void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                pipeline.addLast(new ObjectDecoder(1024*1024, ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
                pipeline.addLast(new ObjectEncoder());
                pipeline.addLast((SimpleChannelInboundHandler)handler);
            }
        });
        serverBootstrap.option(ChannelOption.SO_BACKLOG,1024);
        serverBootstrap.childOption(ChannelOption.SO_KEEPALIVE,true);
        ChannelFuture future = serverBootstrap.bind(address,port).sync();
        //future.channel().closeFuture().sync();
    }finally{
        //workerGroup.shutdownGracefully();
        //bossGroup.shutdownGracefully();
    }
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    pipeline.addLast(new LoggingHandler());
    // Add SSL handler first to encrypt and decrypt everything.
    // In this example, we use a bogus certificate in the server side
    // and accept any invalid certificates in the client side.
    // You will need something more complicated to identify both
    // and server in the real world.
    if (sslCtx != null)
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), SecureChatClient.HOST, SecureChatClient.PORT));

    // On top of the SSL handler, add the text line codec.
    pipeline.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
    pipeline.addLast(new StringDecoder());
    pipeline.addLast(new StringEncoder());

    // and then business logic.
    pipeline.addLast(new SecureChatClientHandler());
}

/**
 * Initialize the {@code SocketChannel}.
 *
 * This method initializes a new channel created by the {@code ServerBootstrap}
 *
 * The default implementation create a remote connection, configures a default pipeline
 * which handles coding/decoding messages, handshaking, timeout and error handling based
 * on {@code RpcConfig} instance provided at construction time.
 *
 * Subclasses can override it to add extra handlers if needed.
 *
 * Note that this method might be called while the instance is still under construction.
 *
 * @param ch the socket channel
 */
protected void initChannel(final SocketChannel ch) {
  C connection = initRemoteConnection(ch);
  connection.setChannelCloseHandler(getCloseHandler(ch, connection));

  final ChannelPipeline pipeline = ch.pipeline();
  pipeline.addLast(PROTOCOL_ENCODER, new RpcEncoder("s-" + rpcConfig.getName()));
  pipeline.addLast("message-decoder", getDecoder(connection.getAllocator()));
  pipeline.addLast("handshake-handler", getHandshakeHandler(connection));

  if (rpcConfig.hasTimeout()) {
    pipeline.addLast(TIMEOUT_HANDLER,
        new LogggingReadTimeoutHandler(connection, rpcConfig.getTimeout()));
  }

  pipeline.addLast("message-handler", new InboundHandler(connection));
  pipeline.addLast("exception-handler", new RpcExceptionHandler<>(connection));
}

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

public void run() throws Exception {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .localAddress(new InetSocketAddress(port))
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new EchoServerHandler());
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true);
        ChannelFuture f = b.bind(port).sync();
        f.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

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

private static CompletableFuture<Void> closeChannelGroup(
    ChannelGroup channelGroup, CloseType closeType) {
  switch (closeType) {
    case DISCONNECT:
      return completable(channelGroup.disconnect());
    default:
      return CompletableFuture.allOf(
          channelGroup
              .stream()
              .map(
                  c -> {
                    CompletableFuture<Void> f;
                    Function<SocketChannel, ChannelFuture> shutdownMethod =
                        closeType == CloseType.SHUTDOWN_READ
                            ? SocketChannel::shutdownInput
                            : SocketChannel::shutdownOutput;
                    if (c instanceof SocketChannel) {
                      f = completable(shutdownMethod.apply((SocketChannel) c));
                    } else {
                      logger.warn(
                          "Got {} request for non-SocketChannel {}, disconnecting instead.",
                          closeType,
                          c);
                      f = completable(c.disconnect());
                    }
                    return f;
                  })
              .collect(Collectors.toList())
              .toArray(new CompletableFuture[] {}));
  }
}

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    channel.pipeline()
            .addLast(new ReadTimeoutHandler(30))
            .addLast("splitter", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4))
            .addLast(new PacketDecoder())
            .addLast("prepender", new LengthFieldPrepender(4))
            .addLast(new PacketEncoder())
            .addLast(client.getHandler());
    this.client.setChannel(channel);
    System.out.println("Netty client started");
}

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

/**
 *@description 连接服务器
 *@time 创建时间:2017年7月21日下午4:15:50
 *@param host
 *@param port
 *@throws InterruptedException
 *@author dzn
 */
public void connect(String host, int port) throws InterruptedException{
    EventLoopGroup group = new NioEventLoopGroup();
    try{
        Bootstrap boot = new Bootstrap();
        boot.group(group)
        .channel(NioSocketChannel.class)
        .option(ChannelOption.TCP_NODELAY, true)
        .handler(new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                //增加以\n 和 \r\n为数据换行符的Handler
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                //增加字符串解析器
                ch.pipeline().addLast(new StringDecoder());
                //对输入数据进行业务逻辑处理
                ch.pipeline().addLast(new RightTimeClientHandler());
            }

        });

        //连接服务器
        ChannelFuture future = boot.connect(host, port).sync();

        //等待客户端Channel关闭
        future.channel().closeFuture().sync();

    }finally{
        group.shutdownGracefully();
    }
}

@Override
public void start() {
    b.group(workGroup)
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast("SdkServerDecoder", new SdkClientDecoder(12))
                            .addLast("SdkServerEncoder", new SdkClientEncoder())
                            .addLast("SdkClientHandler", new SdkClientHandler());
                }
            });
    try {
        cf = b.connect(GlobalConfig.DEFAULT_HOST, GlobalConfig.SDKS_PORT).sync();
        cf.channel().closeFuture().addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture channelFuture) throws Exception {
                logger.error("client channel close", channelFuture.cause());
                shutdown();
            }
        });

        InetSocketAddress address = (InetSocketAddress) cf.channel().remoteAddress();
        logger.info("SdkClient start success, host is {}, port is {}", address.getHostName(),
                address.getPort());
    } catch (InterruptedException e) {
        logger.error("SdkClient start error", e);
        shutdown(); //关闭并释放资源
    }
}

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

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

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();
    if (sslCtx != null) {
        pipeline.addLast("ssl-handler", sslCtx.newHandler(ch.alloc()));
    }
    //pipeline.addLast("http-compressor", new HttpContentCompressor());
    pipeline.addLast("http-codec", new HttpServerCodec());
    pipeline.addLast("http-aggregator", new HttpObjectAggregator(65536));
    pipeline.addLast("http-chunked", new ChunkedWriteHandler());
    pipeline.addLast("http-handler", new HttpFileServerHandler());
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {

    ChannelPipeline pipeline = socketChannel.pipeline();
    pipeline.addLast(new HttpServerCodec());
    pipeline.addLast(new HttpObjectAggregator(65536));
    pipeline.addLast(new ChunkedWriteHandler());
    pipeline.addLast(new DefaultHttpServerHandler(ahsc));
}

public void start() throws InterruptedException {
    ServerBootstrap b = new ServerBootstrap();
    b.option(ChannelOption.SO_LINGER, socketLinger);
    b.option(ChannelOption.SO_REUSEADDR, reuseAddress);
    b.group(bossGroup, workerGroup)
        .channel(NioServerSocketChannel.class)
        .handler(new LoggingHandler(LogLevel.INFO))
        .childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline p = ch.pipeline();
                p.addLast(new HttpRequestDecoder());
                p.addLast(new HttpObjectAggregator(1024 * 1024 * 5));
                p.addLast(new HttpResponseEncoder());
                p.addLast(new HttpContentCompressor());
                if (corsConfiguration.hasHeader()) {
                    p.addLast(new CorsHandler(
                        CorsConfig
                            .withOrigin(corsConfiguration.getHeader())
                            .allowedRequestHeaders(HttpHeaders.Names.CONTENT_TYPE)
                            .allowedRequestMethods(HttpMethod.POST)
                        .build())
                    );
                }
                p.addLast(jsonRpcWeb3FilterHandler);
                p.addLast(jsonRpcWeb3ServerHandler);
            }
        });
    b.bind(host, port).sync();
}

public static void main(String[] args) throws Exception {
    // Configure SSL.git
    final SslContext sslCtx;
    if (SSL) {
        sslCtx = SslContextBuilder.forClient()
            .trustManager(InsecureTrustManagerFactory.INSTANCE).build();
    } else {
        sslCtx = null;
    }

    // Configure the client.
    EventLoopGroup group = new NioEventLoopGroup();
    try {
        Bootstrap b = new Bootstrap();
        b.group(group)
         .channel(NioSocketChannel.class)
         .option(ChannelOption.TCP_NODELAY, true)
         .handler(new ChannelInitializer<SocketChannel>() {
             @Override
             public void initChannel(SocketChannel ch) throws Exception {
                 ChannelPipeline p = ch.pipeline();
                 if (sslCtx != null) {
                     p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
                 }
                // p.addLast(new LoggingHandler(LogLevel.INFO));
                 p.addLast(new EchoClientHandler());
             }
         });

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

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline pipeline = ch.pipeline();

    if (sslCtx != null) {
        pipeline.addLast(sslCtx.newHandler(ch.alloc(), ip, port));
    }

    pipeline.addLast(DECODER);
    pipeline.addLast(ENCODER);
    // and then business logic.
    pipeline.addLast(CLIENT_HANDLER);

}

protected void initChannel(SocketChannel channel) throws Exception {
    log.info("New channel created");
    channel.pipeline().addLast(new StringEncoder(CharsetUtil.UTF_8));
    channel.pipeline().addLast(new MessageDecoder());
    handleNewNodeConnection(channel);

}

public RemoteConnection(SocketChannel channel, String name, boolean blockOnSocket) {
  super();
  this.channel = channel;
  this.clientName = name;
  this.writeManager = new WriteManager();
  this.requestIdMap = new RequestIdMap(getName());
  if(!blockOnSocket){
    writeManager.disable();
  }
  channel.pipeline().addLast(new BackPressureHandler());
}

public void start() {
    Configuration config = Configuration.INSTANCE;
    InternalLoggerFactory.setDefaultFactory(Slf4JLoggerFactory.INSTANCE);
    EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    EventLoopGroup workerGroup = new NioEventLoopGroup();
    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                .addLast("logging", new LoggingHandler(LogLevel.DEBUG))
                                .addLast(new XConnectHandler());
                        if (config.getReadLimit() != 0 || config.getWriteLimit() != 0) {
                            socketChannel.pipeline().addLast(
                                    new GlobalTrafficShapingHandler(Executors.newScheduledThreadPool(1), config.getWriteLimit(), config.getReadLimit())
                            );
                        }
                    }
                });
        log.info("\tStartup {}-{}-server [{}]", Constants.APP_NAME, Constants.APP_VERSION, config.getProtocol());
        new Thread(() -> new UdpServer().start()).start();
        ChannelFuture future = bootstrap.bind(config.getHost(), config.getPort()).sync();
        future.addListener(future1 -> log.info("\tTCP listening at {}:{}...", config.getHost(), config.getPort()));
        future.channel().closeFuture().sync();
    } catch (Exception e) {
        log.error("\tSocket bind failure ({})", e.getMessage());
    } finally {
        log.info("\tShutting down and recycling...");
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
        Configuration.shutdownRelays();
    }
    System.exit(0);
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {

    ChannelPipeline p = ch.pipeline();

    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast(new HttpClientCodec());
}

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

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG, 0)
                    .handler(new LoggingHandler(LogLevel.DEBUG))
                    .childHandler(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 ServerHandler()
                            );
                        }
                    });

            serverChannel = b.bind(this.port).sync().channel();
        } finally {

        }
    }

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 startSocket() throws InterruptedException {
    EventLoopGroup boss = new NioEventLoopGroup();
    EventLoopGroup worker = new NioEventLoopGroup();

    try {
        ServerBootstrap boot = new ServerBootstrap();
        boot.group(boss,worker);

        boot.channel(NioServerSocketChannel.class);
        boot.childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024*1024,0,4,-4,0,false));
                ch.pipeline().addLast(new ByteToPacketCodec());
                //ch.pipeline().addLast(new LoginChannelHandler(listener));
                ch.pipeline().addLast(new PacketChannelHandler(listener));
            }
        });

        boot.option(ChannelOption.SO_BACKLOG,128);
        boot.childOption(ChannelOption.SO_KEEPALIVE,true);
        channelFuture = boot.bind(port).sync();
        System.out.println("服务器"+port+"开启成功...");
        channelFuture.channel().closeFuture().sync();
    }finally {
        boss.shutdownGracefully().sync();
        worker.shutdownGracefully().sync();
        channelFuture = null;
        System.out.println("服务器关闭成功...");
    }
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline cp = socketChannel.pipeline();
    cp.addLast(new HttpServerCodec());  //添加服务端http编、解码器
    cp.addLast(new HttpObjectAggregator(512*1024));  //http消息聚合
    cp.addLast(new HttpContentCompressor());   //开启压缩
    cp.addLast(new HttpServerHandler(kurdran));
}

@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
    socketChannel.pipeline().addLast(
            new HttpServerCodec(),
            new HttpServerExpectContinueHandler(),
            new HttpObjectAggregator(Integer.MAX_VALUE),
            new ChunkedWriteHandler(),
            new HttpRequestHandler()
    );
}

@Override
protected void initChannel(final SocketChannel socketChannel) {
    final ChannelPipeline pipeline = socketChannel.pipeline();

    pipeline.addLast(new DelimiterBasedFrameDecoder(1048576 * 2, Delimiters.lineDelimiter()));
    pipeline.addLast(new StringDecoder());
    pipeline.addLast(new StringEncoder());

    pipeline.addLast(new ServerHandler());
}

/**
 * The Method that will initialize the channel.
 *
 * @param socketChannel The channel.
 *
 * @throws Exception Codec exception.
 */
protected void initChannel(SocketChannel socketChannel) throws Exception {
    ChannelPipeline pipeline = socketChannel.pipeline();

    pipeline.addLast(new HTTPDecoder());
    pipeline.addLast(new HTTPEncoder());
    pipeline.addLast(new EchidnaConnection(socketChannel, server));
}

public Client ( final SocketAddress address, final ConnectionStateListener listener, final ProtocolOptions options, final List<ClientModule> modules )
{
    this.address = address;
    this.options = options;

    this.listener = listener;

    this.manager = new MessageManager ( options );

    this.group = new NioEventLoopGroup ();

    this.bootstrap = new Bootstrap ();
    this.bootstrap.group ( this.group );
    this.bootstrap.channel ( NioSocketChannel.class );

    this.bootstrap.handler ( new ChannelInitializer<SocketChannel> () {

        @Override
        protected void initChannel ( final SocketChannel ch ) throws Exception
        {
            handleInitChannel ( ch );
        }
    } );

    this.modules = modules.toArray ( new ClientModule[modules.size ()] );
    this.executor = Executors.newSingleThreadExecutor ( new NamedThreadFactory ( "IEC60870Client/" + address ) );

    for ( final ClientModule module : modules )
    {
        module.initializeClient ( this, this.manager );
    }
}

@Before
public void setUp()
throws Exception {

    serverMock = new NioConnDroppingServer(DEFAULT_PORT, FAIL_EVERY_CONN_ATTEMPT);

    final Semaphore concurrencyThrottle = new Semaphore(CONCURRENCY);
    group = new NioEventLoopGroup();
    final Bootstrap bootstrap = new Bootstrap()
        .group(group)
        .channel(NioSocketChannel.class)
        .handler(
            new ChannelInitializer<SocketChannel>() {
                @Override
                protected final void initChannel(final SocketChannel conn)
                throws Exception {
                    conn.pipeline().addLast(new DummyClientChannelHandler());
                }
            }
        )
        .option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.SO_REUSEADDR, true)
        .option(ChannelOption.TCP_NODELAY, true);
    connPool = new BasicMultiNodeConnPool(
        concurrencyThrottle, NODES, bootstrap, CPH, DEFAULT_PORT, 0
    );
    connPool.preCreateConnections(CONCURRENCY);
}

@Override
public void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("encoder", new MqttMessageEncoder());
    pipeline.addLast("decoder", new MqttMessageDecoder());
    pipeline.addLast("handler", new MqttMessageHandler());
}

protected ChannelHandler setupHttpChannel(Configuration config, SslContext sslCtx) {

        return new ChannelInitializer<SocketChannel>() {

            @Override
            protected void initChannel(SocketChannel ch) throws Exception {

                ch.pipeline().addLast("ssl", new NonSslRedirectHandler(config, sslCtx));
                ch.pipeline().addLast("encoder", new HttpResponseEncoder());
                ch.pipeline().addLast("decoder", new HttpRequestDecoder());
                ch.pipeline().addLast("compressor", new HttpContentCompressor());
                ch.pipeline().addLast("decompressor", new HttpContentDecompressor());
                ch.pipeline().addLast("aggregator", new HttpObjectAggregator(8192));
                ch.pipeline().addLast("chunker", new ChunkedWriteHandler());
                final Configuration.Cors corsCfg = config.getHttp().getCors();
                final CorsConfig.Builder ccb;
                if (corsCfg.isAllowAnyOrigin()) {
                    ccb = new CorsConfig.Builder();
                } else {
                    ccb = new CorsConfig.Builder(corsCfg.getAllowedOrigins().stream().toArray(String[]::new));
                }
                if (corsCfg.isAllowNullOrigin()) {
                    ccb.allowNullOrigin();
                }
                if (corsCfg.isAllowCredentials()) {
                    ccb.allowCredentials();
                }
                corsCfg.getAllowedMethods().stream().map(HttpMethod::valueOf).forEach(ccb::allowedRequestMethods);
                corsCfg.getAllowedHeaders().forEach(ccb::allowedRequestHeaders);
                CorsConfig cors = ccb.build();
                LOG.trace("Cors configuration: {}", cors);
                ch.pipeline().addLast("cors", new CorsHandler(cors));
                ch.pipeline().addLast("queryDecoder", new qonduit.netty.http.HttpRequestDecoder(config));
                ch.pipeline().addLast("strict", new StrictTransportHandler(config));
                ch.pipeline().addLast("login", new X509LoginRequestHandler(config));
                ch.pipeline().addLast("doLogin", new BasicAuthLoginRequestHandler(config));
                ch.pipeline().addLast("error", new HttpExceptionHandler());
            }
        };
    }

protected void initChannel(SocketChannel socketChannel) {
    socketChannel.pipeline()
                 .addLast(new AmqpDecoder())
                 .addLast(new AmqpEncoder())
                 .addLast(new AmqpConnectionHandler(configuration, broker))
                 .addLast(ioExecutors, new AmqpMessageWriter())
                 .addLast(ioExecutors, new BlockingTaskHandler());
}