Java 类io.netty.handler.codec.LengthFieldPrepender 实例源码

public void run() {
    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 {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65536, 0, 4, 0, 4));
                pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                pipeline.addLast("decoder", new MsgPackDecode());
                pipeline.addLast("encoder", new MsgPackEncode());
                pipeline.addLast(new ClientHandler());
            }
        });
        channel = b.connect(clientProperties.getServerHost(), clientProperties.getServerPort()).sync().channel();
        status = Status.START;
        channel.closeFuture().sync();
    } catch (Exception e) {
        e.printStackTrace();
    }
    status = Status.STOP;
}

@Override
protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pip = ch.pipeline();
    int maxLength = 1048576;
    int lengthFieldLength = 4;
    int ignoreLength = -4;
    int offset = 0;
    pip.addLast(new LengthFieldBasedFrameDecoder(maxLength, offset, lengthFieldLength, ignoreLength, lengthFieldLength));
    pip.addLast(new MessageDecoder(builder.getImessageandhandler()));
    pip.addLast(new LengthFieldPrepender(4, true));
    pip.addLast(new MessageEncoder(builder.getImessageandhandler()));
    pip.addLast(new MessageExecutor(builder.getConsumer(), builder.getListener()));
    for (ChannelHandler handler : builder.getExtraHandlers()) {
        pip.addLast(handler);
    }
}

public static void service() throws Exception {
    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workerGroup);
    bootstrap.channel(NioServerSocketChannel.class);
    bootstrap.childHandler(new ChannelInitializer<Channel>() {

        @Override
        protected void initChannel(Channel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast(new LengthFieldPrepender(4));
            pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast(new TcpServerHandler());
        }

    });
    ChannelFuture f = bootstrap.bind(IP, PORT).sync();
    f.channel().closeFuture().sync();
    System.out.println("TCP服务器已启动");
}

/**
 * 初始化Bootstrap
 * @return
 */
public static final Bootstrap getBootstrap(){
    EventLoopGroup group = new NioEventLoopGroup();
    Bootstrap b = new Bootstrap();
    b.group(group).channel(NioSocketChannel.class);
    b.handler(new ChannelInitializer<Channel>() {
        @Override
        protected void initChannel(Channel 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("handler", new TcpClientHandler());
        }
    });
    b.option(ChannelOption.SO_KEEPALIVE, true);
    return b;
}

protected static void run() throws Exception {
    ServerBootstrap b = new ServerBootstrap();
    b.group(bossGroup, workerGroup);
    b.channel(NioServerSocketChannel.class);
    b.childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public 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 TcpServerHandler());
        }
    });

    b.bind(IP, PORT).sync();
    System.out.println("TCP服务器已启动");
}

@Provides
@EppProtocol
static ImmutableList<Provider<? extends ChannelHandler>> provideHandlerProviders(
    Provider<SslServerInitializer<NioSocketChannel>> sslServerInitializerProvider,
    Provider<ProxyProtocolHandler> proxyProtocolHandlerProvider,
    @EppProtocol Provider<ReadTimeoutHandler> readTimeoutHandlerProvider,
    Provider<LengthFieldBasedFrameDecoder> lengthFieldBasedFrameDecoderProvider,
    Provider<LengthFieldPrepender> lengthFieldPrependerProvider,
    Provider<EppServiceHandler> eppServiceHandlerProvider,
    Provider<LoggingHandler> loggingHandlerProvider,
    Provider<FullHttpRequestRelayHandler> relayHandlerProvider) {
  return ImmutableList.of(
      proxyProtocolHandlerProvider,
      sslServerInitializerProvider,
      readTimeoutHandlerProvider,
      lengthFieldBasedFrameDecoderProvider,
      lengthFieldPrependerProvider,
      eppServiceHandlerProvider,
      loggingHandlerProvider,
      relayHandlerProvider);
}

@Test
public void testPrependLengthInLittleEndian() throws Exception {
    final EmbeddedChannel ch = new EmbeddedChannel(new LengthFieldPrepender(ByteOrder.LITTLE_ENDIAN, 4, 0, false));
    ch.writeOutbound(msg);
    ByteBuf buf = (ByteBuf) ch.readOutbound();
    assertEquals(5, buf.readableBytes());
    byte[] writtenBytes = new byte[buf.readableBytes()];
    buf.getBytes(0, writtenBytes);
    assertEquals(1, writtenBytes[0]);
    assertEquals(0, writtenBytes[1]);
    assertEquals(0, writtenBytes[2]);
    assertEquals(0, writtenBytes[3]);
    assertEquals('A', writtenBytes[4]);
    buf.release();
    assertFalse("The channel must have been completely read", ch.finish());
}

public void connect() throws Exception {
    workerGroup = new NioEventLoopGroup();

    Bootstrap bootstrap =
            new Bootstrap()
                    .group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();

                            pipeline.addLast(new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
                            pipeline.addLast(new ProtobufDecoder(Protocol.BaseMessage.getDefaultInstance()));

                            clientHandler = new ClientHandler();
                            pipeline.addLast(clientHandler);

                            pipeline.addLast(new LengthFieldPrepender(4));
                            pipeline.addLast(new ProtobufEncoder());
                        }
                    });

    ChannelFuture channelFuture = bootstrap.connect(host, port).sync();
    channel = channelFuture.channel();
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  LOGGER.info("Setting up Server channel !!");
  ch.pipeline().addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  ch.pipeline().addLast("encoder", new LengthFieldPrepender(4));
  //ch.pipeline().addLast("logger", new LoggingHandler());
  // Create server metric for this handler and add to aggregate if present
  NettyServerMetrics serverMetric =
      new NettyServerMetrics(_registry, NettyTCPServer.class.getName() + "_" + Utils.getUniqueId() + "_");

  if (null != _globalMetrics) {
    _globalMetrics.addTransportClientMetrics(serverMetric);
  }

  ch.pipeline().addLast("request_handler",
      new NettyChannelInboundHandler(_handlerFactory.createNewRequestHandler(), serverMetric, _defaultLargeQueryLatencyMs));
}

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

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

    // In
    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
    if (cryptoFunction != null) {
        ch.pipeline().addLast(cryptoFunction.getDecoder());
    }
    ch.pipeline().addLast(new PacketDecoder(protocol));

    // Out
    ch.pipeline().addLast(new LengthFieldPrepender(4));
    if (cryptoFunction != null) {
        ch.pipeline().addLast(cryptoFunction.getEncoder());
    }
    ch.pipeline().addLast(new PacketEncoder(protocol));

    // Handler
    ch.pipeline().addLast(new CascadeSession(ch, protocol, sessionListener));
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // Enable TCPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast("logger", new LoggingHandler(LogLevel.DEBUG));

    p.addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, Integer.MAX_VALUE, 0, 4, 0, 4, true));
    p.addLast(new ByteBufferDecoder());

    p.addLast(new LengthFieldPrepender(4, 0, false) {
        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
            ByteBuf outWithLittleEndian = out.order(ByteOrder.LITTLE_ENDIAN);
            super.encode(ctx, msg, outWithLittleEndian);
        }
    });
    p.addLast(new ByteBufferEncoder());
    p.addLast(handler);
}

@Override
public void initChannel(SocketChannel ch) {
    ChannelPipeline p = ch.pipeline();
    // Enable TCPS if necessary.
    if (sslCtx != null) {
        p.addLast(sslCtx.newHandler(ch.alloc()));
    }
    p.addLast("logger", new LoggingHandler(LogLevel.DEBUG));

    p.addLast(new LengthFieldBasedFrameDecoder(ByteOrder.LITTLE_ENDIAN, Integer.MAX_VALUE, 0, 4, 0, 4, true));
    p.addLast(new ByteBufferDecoder());

    p.addLast(new LengthFieldPrepender(4, 0, false) {
        @Override
        protected void encode(ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
            ByteBuf outWithLittleEndian = out.order(ByteOrder.LITTLE_ENDIAN);
            super.encode(ctx, msg, outWithLittleEndian);
        }
    });
    p.addLast(new ByteBufferEncoder());
    p.addLast(new TcpHandler(transportFactory, path, connectionListener));
}

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

    // 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.

    SSLEngine engine =
        SecureSocketSslContextFactory.getClientContext().createSSLEngine();
    engine.setUseClientMode(true);

    pipeline.addLast("ssl", new SslHandler(engine));
    // On top of the SSL handler, add the text line codec.
    // pipeline.addLast("framer", new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
    pipeline.addLast("length-decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    pipeline.addLast("String-decoder", new StringDecoder());  
    pipeline.addLast("length-encoder", new LengthFieldPrepender(4));
    pipeline.addLast("String-encoder", new StringEncoder()); 
    pipeline.addLast("handler", new SecureSocketClientHandler());
}

public void run() throws InterruptedException {
    EventLoopGroup bossGroup = new NioEventLoopGroup();
    EventLoopGroup workerGroup = new NioEventLoopGroup();

    try {
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup)
                .localAddress(port)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LengthFieldPrepender(4)).addLast(new LengthFieldServerHandler());
                    }
                });

        ChannelFuture cf = bootstrap.bind().sync();
        cf.channel().closeFuture().sync();
    } finally {
        workerGroup.shutdownGracefully();
        bossGroup.shutdownGracefully();
    }
}

public void run(String host, int port) throws Exception {

        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG))
                                    .addLast(new LengthFieldPrepender(4))
                                    .addLast(new StringEncoder())
                                    .addLast(new ObjectToJsonStringEncoder())
                                    .addLast(new ObjectEchoClientHandler());
                        }
                    });

            bootstrap.connect(host, port).sync().channel().closeFuture().sync();
        } finally {
            group.shutdownGracefully();
        }
    }

public TCPServerHandler(EventLoopGroup bossGroup, EventLoopGroup workerGroup, ThrowableNioEventLoopGroup eventExecutor, List<Service> services, InetSocketAddress serverAddress) throws InterruptedException {
    ChannelFuture bind = new ServerBootstrap()
            .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
            .group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .option(ChannelOption.AUTO_READ, false)
            .option(ChannelOption.SO_BACKLOG, 100)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ChannelPipeline p = ch.pipeline();
                    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));
                    p.addLast("packetDecoder", new PacketDecoder());
                    p.addLast("frameEncoder", new LengthFieldPrepender(Integer.BYTES));
                    p.addLast("packetEncoder", new PacketEncoder());
                    p.addLast(new ServerChannelAdapter(services, eventExecutor));
                }
            }).bind(serverAddress);

    server = bind.sync()
            .addListener(future -> {
                if (!future.isSuccess()) {
                    LOGGER.error("Failed to bind {}", bind.channel().localAddress());
                }
            }).awaitUninterruptibly().channel();
}

public Sender(final String source, final String destination) {
  this.destination = destination;
  bootstrap = new Bootstrap();
  bootstrap.group(workerGroup);
  bootstrap.channel(NioSocketChannel.class);
  bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 1000);
  bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
  bootstrap.option(ChannelOption.TCP_NODELAY, true);
  bootstrap.handler(new ChannelInitializer<SocketChannel>() {
    @Override
    public void initChannel(SocketChannel ch) throws Exception {
      if (isSslEnabled()) {
        SSLEngine engine = serverContext.createSSLEngine();
        engine.setUseClientMode(true);
        ch.pipeline().addLast(new SslHandler(engine));
      }
      // Inbound handlers.
      ch.pipeline().addLast("clientError", new ClientErrorHandler());
      // Outbound handlers.
      ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
    }
  });
}

@Override
protected void initChannel(SocketChannel ch) throws Exception {
  LOGGER.info("Setting up Server channel, scheduler");
  ch.pipeline().addLast("decoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
  ch.pipeline().addLast("encoder", new LengthFieldPrepender(4));
  //ch.pipeline().addLast("logger", new LoggingHandler());
  // Create server metric for this handler and add to aggregate if present
  NettyServerMetrics serverMetric =
      new NettyServerMetrics(_registry, NettyTCPServer.class.getName() + "_" + Utils.getUniqueId() + "_");

  if (null != _globalMetrics) {
    _globalMetrics.addTransportClientMetrics(serverMetric);
  }

  ch.pipeline().addLast("request_handler",
      new NettyChannelInboundHandler(_handlerFactory.createNewRequestHandler(), serverMetric, _defaultLargeQueryLatencyMs));
}

/**
 * Initializes a new ClientChannelInitializer.
 * 
 * @param clientChannelManager
 *            a ClientChannelManager that will be used by the {@link HelloMessageHandler}, the
 *            {@link ConnectionEstablishedMessageHandler} and for registering new Channels
 * @param localID
 *            the ModuleID of the client this initializer is running on
 */
public ClientChannelInitializer(final ClientChannelManager clientChannelManager, final ModuleID localID) {
    this.clientChannelManager = clientChannelManager;

    final List<ChannelHandler> handlers = new ArrayList<ChannelHandler>();

    handlers.add(new LengthFieldPrepender(LENGTH_FIELD_LENGTH));

    final Charset charset = Charset.forName("UTF-8");
    handlers.add(new StringEncoder(charset));
    handlers.add(new StringDecoder(charset));

    messageAdapter.addMessageType(HelloMessage.class);
    messageAdapter.addMessageType(ConnectionEstablishedMessage.class);
    gsonCodec.registerTypeAdapter(Message.class, messageAdapter);
    handlers.add(gsonCodec);

    handlers.add(new HelloMessageHandler(clientChannelManager, localID));
    handlers.add(new ConnectionEstablishedMessageHandler(clientChannelManager, localID));

    defaultHandlers = Collections.unmodifiableList(handlers);

    firstMessage = new HelloMessage(localID, MAX_FRAME_LENGTH);
}

private static <T> void serveMany(int port, final Observable<List<Observable<T>>> observable, final Encoder<T> encoder,
            boolean startAndWait, SlottingStrategy<T> slottingStrategy, IngressPolicy ingressPolicy){
        RxServer<RemoteRxEvent, RemoteRxEvent> server 
            = RxNetty.createTcpServer(port, new PipelineConfiguratorComposite<RemoteRxEvent, RemoteRxEvent>(
                new PipelineConfigurator<RemoteRxEvent, RemoteRxEvent>(){
                    @Override
                    public void configureNewPipeline(ChannelPipeline pipeline) {
//                      pipeline.addFirst(new LoggingHandler(LogLevel.ERROR)); // uncomment to enable debug logging 
                        pipeline.addLast("frameEncoder", new LengthFieldPrepender(4)); // 4 bytes to encode length
                        pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(524288, 0, 4, 0, 4)); // max frame = half MB
                    }
                }, new RxEventPipelineConfigurator()),  
                new RemoteObservableConnectionHandler<T>(observable, encoder, slottingStrategy, ingressPolicy));
        if(startAndWait){
            server.startAndWait();
        }else{
            server.start();
        }
    }

/**
 * Returns a new channel initializer suited to encode and decode a protocol
 * buffer message.
 * <p/>
 * <p>Message sizes over 10 MB are not supported.</p>
 * <p/>
 * <p>The handler will be executed on the I/O thread. Blocking operations
 * should be executed in their own thread.</p>
 *
 * @param defaultInstance an instance of the message to handle
 * @param handler the handler implementing the application logic
 * @param <M> the type of the support protocol buffer message
 */
public static final <M extends Message> ChannelInitializer<Channel> protoBuf(
    final M defaultInstance, final SimpleChannelInboundHandler<M> handler) {
  return new ChannelInitializer<Channel>() {

    @Override
    protected void initChannel(Channel channel) throws Exception {
      channel.pipeline().addLast("frameDecoder",
          new LengthFieldBasedFrameDecoder(10 * 1024 * 1024, 0, 4, 0, 4));
      channel.pipeline().addLast("protobufDecoder",
          new ProtobufDecoder(defaultInstance));
      channel.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
      channel.pipeline().addLast("protobufEncoder", new ProtobufEncoder());
      channel.pipeline().addLast("applicationHandler", handler);
    }
  };
}

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

public void start() {
  Configuration conf = context.getConf();
  int workerNum =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_EVENTGROUP_THREADNUM,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_EVENTGROUP_THREADNUM);

  int sendBuffSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_SNDBUF,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_SNDBUF);

  int recvBuffSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_SERVER_RCVBUF,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_SERVER_RCVBUF);

  final int maxMessageSize =
      conf.getInt(AngelConf.ANGEL_NETTY_MATRIXTRANSFER_MAX_MESSAGE_SIZE,
          AngelConf.DEFAULT_ANGEL_NETTY_MATRIXTRANSFER_MAX_MESSAGE_SIZE);

  bossGroup = new NioEventLoopGroup(1);
  workerGroup = new NioEventLoopGroup(workerNum);
  ((NioEventLoopGroup) workerGroup).setIoRatio(70);

  ServerBootstrap b = new ServerBootstrap();
  b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
      .option(ChannelOption.SO_SNDBUF, sendBuffSize)
      .option(ChannelOption.SO_RCVBUF, recvBuffSize)
      .childHandler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
          ChannelPipeline p = ch.pipeline();
          p.addLast(new LengthFieldBasedFrameDecoder(maxMessageSize, 0, 4, 0, 4));
          p.addLast(new LengthFieldPrepender(4));
          p.addLast(new MatrixTransportServerHandler(context));
        }
      });

  channelFuture = b.bind(port);
}

@Override
protected void initChannel(T ch) throws Exception {
    ChannelPipeline p = ch.pipeline();

    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    p.addLast("protobufDecoder", new ProtobufDecoder(NettyRpcProto.RpcContainer.getDefaultInstance()));

    p.addLast("frameEncoder", new LengthFieldPrepender(4));
    p.addLast("protobufEncoder", new ProtobufEncoder());

    ConcurrentHashMap<Integer, RpcCall> callMap = new ConcurrentHashMap<Integer, RpcCall>();
    p.addLast(eventExecutor, "inboundHandler", new InboundHandler(callMap));
    p.addLast("outboundHandler", new OutboundHandler(callMap));

}

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

    p.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
    p.addLast("protobufDecoder", new ProtobufDecoder(NettyRpcProto.RpcContainer.getDefaultInstance()));

    p.addLast("frameEncoder", new LengthFieldPrepender(4));
    p.addLast("protobufEncoder", new ProtobufEncoder());

    p.addLast(eventExecutor, "serverHandler", handler);
}

@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 start() {
    int heartTime = transaction_netty_heart_time+10;
    txCoreServerHandler = new TxCoreServerHandler(mqTxManagerService);
    bossGroup = new NioEventLoopGroup(50); // (1)
    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 ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast("timeout", new IdleStateHandler(heartTime, heartTime, heartTime, TimeUnit.SECONDS));

                        ch.pipeline().addLast(new LengthFieldPrepender(4, false));
                        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));

                        ch.pipeline().addLast(txCoreServerHandler);
                    }
                });

        // Start the server.
        b.bind(Constants.socketPort);
        logger.info("Socket started on port(s): " + Constants.socketPort + " (socket)");

    } catch (Exception e) {
        // Shut down all event loops to terminate all threads.
        e.printStackTrace();
    }
}

private ChannelInitializer channelInitializer() {
    return new ChannelInitializer<SocketChannel>() {
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            ChannelPipeline pipeline = socketChannel.pipeline();
            pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
            pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
            pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
            pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
            pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
            pipeline.addLast("handler", nodeChannelInBoundHandler());
        }
    };
}

@PostConstruct
public void start() throws Exception {

    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workGroup)
            .channelFactory(NioServerSocketChannel::new)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));
                    pipeline.addLast("handler", nodeChannelInBoundHandler);
                }
            })
            .option(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .option(ChannelOption.SO_SNDBUF, 2048)
            .option(ChannelOption.SO_RCVBUF, 1024);
    bootstrap.bind(goPushNodeServerConfig.getNodePort()).sync();
    log.info("Node server start successful! listening port: {}", goPushNodeServerConfig.getNodePort());
}

@PostConstruct
public void start() throws Exception {


    ServerBootstrap bootstrap = new ServerBootstrap();
    bootstrap.group(bossGroup, workGroup)
            .channelFactory(NioServerSocketChannel::new)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {

                    ChannelPipeline pipeline = socketChannel.pipeline();
                    pipeline.addLast("logHandler", new LoggingHandler());
                    pipeline.addLast("frameDecoder", new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                    pipeline.addLast("stringDecoder", new StringDecoder(CharsetUtil.UTF_8));
                    pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
                    pipeline.addLast("stringEncoder", new StringEncoder(CharsetUtil.UTF_8));
                    pipeline.addLast("idleStateHandler", new IdleStateHandler(300, 0, 0));

                    pipeline.addLast("handler", deviceChannelInboundHandler);
                }
            })

            .option(ChannelOption.SO_BACKLOG, 1000000)  //连接队列深度
            .option(ChannelOption.TCP_NODELAY, true)   //设置 no_delay
            .option(ChannelOption.SO_SNDBUF, 2048).option(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.TCP_NODELAY, true)
            .childOption(ChannelOption.SO_REUSEADDR, true)
            .childOption(ChannelOption.SO_SNDBUF, 2048).childOption(ChannelOption.SO_RCVBUF, 1024)
            .childOption(ChannelOption.SO_LINGER, 0);

    bootstrap.bind(goPushNodeServerConfig.getDevicePort()).sync();
    log.info("device server start successful! listening port: {}", goPushNodeServerConfig.getDevicePort());
}

public void connect() {
    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 {
            ch.pipeline().addFirst(new LoggingHandler(LogLevel.DEBUG));

            ch.pipeline().addLast(new LengthFieldPrepender(4));
            ch.pipeline().addLast(new FrameEncoder());

            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));

            ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                    ByteBuf buf = (ByteBuf) msg;
                    messageHandler.accept(buf);
                }
            });
        }

    });


    ChannelFuture channelFuture = b.connect(host, port).syncUninterruptibly();

    channel = channelFuture.channel();

}

public void run() {
    ServerBootstrap b = new ServerBootstrap();// 引导辅助程序

    bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
    workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);
    try {
        b.group(bossGroup, workerGroup);
        b.channel(NioServerSocketChannel.class);// 设置nio类型的channel
        b.childHandler(new ChannelInitializer<SocketChannel>() {// 有连接到达时会创建一个channel
            protected void initChannel(SocketChannel ch) throws Exception {
                logger.debug("客户端:{} 初始化", ch.remoteAddress());
                // pipeline管理channel中的Handler，在channel队列中添加一个handler来处理业务
                ch.pipeline().addLast("frameDecoder",
                        new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
                ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(4));
                ch.pipeline().addLast("decoder", msgPackDecode);
                ch.pipeline().addLast("encoder", msgPackEncode);
                ch.pipeline().addLast(serverHandler);
            }
        });
        b.option(ChannelOption.SO_BACKLOG, 128);
        b.childOption(ChannelOption.SO_KEEPALIVE, true);
        logger.info("server start : {}", port);
        ChannelFuture f = b.bind(port).sync();// 配置完成，开始绑定server，通过调用sync同步方法阻塞直到绑定成功
        channel = f.channel();
        f.channel().closeFuture().sync();// 应用程序会一直等待，直到channel关闭
    } catch (Exception e) {
        e.printStackTrace();
    }

}

public void initialize(String host, int port) {
    this.host = host;
    this.port = port;
    //
    bootstrap = new Bootstrap();
    if (Epoll.isAvailable()) {
        this.childGroup = new EpollEventLoopGroup(cThreadCount);
        this.bootstrap.group(childGroup).channel(EpollSocketChannel.class);
    } else {
        this.childGroup = new NioEventLoopGroup(cThreadCount);
        this.bootstrap.group(childGroup).channel(NioSocketChannel.class);
    }
    // handlers
    this.prepender = new LengthFieldPrepender(this.lengthFieldLength, false);
    bootstrap.handler(new ChannelInitializer() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ClientContext.this.initChannel(ch);
        }
    });
    //
    this.defaultOptions();
    if (!options.isEmpty()) {
        for (Map.Entry<ChannelOption<Object>, Object> entry : options.entrySet()) {
            bootstrap.option(entry.getKey(), entry.getValue());
        }
    }
}

/**
 * 初始化
 *
 * @param pThreadCount parent thread count.
 * @param cThreadCount worker thread count.
 * @param options      netty network options。
 */
public void initialize(int pThreadCount, int cThreadCount,
                       Map<ChannelOption<Object>, Object> options) {
    this.bootstrap = new ServerBootstrap();
    if (Epoll.isAvailable()) {
        this.parentGroup = new EpollEventLoopGroup(pThreadCount);
        this.childGroup = new EpollEventLoopGroup(cThreadCount);
        this.bootstrap.group(parentGroup, childGroup).channel(EpollServerSocketChannel.class);
    } else {
        this.parentGroup = new NioEventLoopGroup(pThreadCount);
        this.childGroup = new NioEventLoopGroup(cThreadCount);
        this.bootstrap.group(parentGroup, childGroup).channel(NioServerSocketChannel.class);
    }
    // handlers
    this.prepender = new LengthFieldPrepender(this.lengthFieldLength, false);
    bootstrap.childHandler(new ChannelInitializer() {
        @Override
        protected void initChannel(Channel ch) throws Exception {
            ServerContext.this.initChannel(ch);
        }
    });
    //
    this.defaultOptions();
    if (!options.isEmpty()) {
        for (Map.Entry<ChannelOption<Object>, Object> entry : options.entrySet()) {
            bootstrap.childOption(entry.getKey(), entry.getValue());
        }
    }
}

@Override
protected void initChannel(Channel channel) throws Exception {
    try {
        channel.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException e) {
        // Not supported
    }
    channel.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
    channel.pipeline().addLast(new JSONObjectDecoder());
    channel.pipeline().addLast(new LengthFieldPrepender(4));
    channel.pipeline().addLast(new JSONObjectEncoder());
    channel.pipeline().addLast(nioSocketClient);
}

@Override
protected void initChannel(Channel channel) throws Exception {
    try {
        channel.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException e) {
        // Not supported
    }
    channel.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
    channel.pipeline().addLast(new JSONObjectDecoder());
    channel.pipeline().addLast(new LengthFieldPrepender(4));
    channel.pipeline().addLast(new JSONObjectEncoder());
    channel.pipeline().addLast(new Connection(jaPSServer, channel));
}

@Override
protected void initChannel(Channel channel) throws Exception {
    try {
        channel.config().setOption(ChannelOption.IP_TOS, 0x18);
    } catch (ChannelException e) {
        // Not supported
    }
    channel.config().setAllocator(PooledByteBufAllocator.DEFAULT);

    channel.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4));
    channel.pipeline().addLast(new JSONObjectDecoder());
    channel.pipeline().addLast(new LengthFieldPrepender(4));
    channel.pipeline().addLast(new JSONObjectEncoder());
    channel.pipeline().addLast(clusterPublisher);
}

public void start(){
    b.group(group).channel(NioSocketChannel.class)
            .option(ChannelOption.TCP_NODELAY, true)
            .option(ChannelOption.SO_KEEPALIVE, true)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch)
                        throws Exception {
                    ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(1<<20, 0, 4, 0, 4),
                            new LengthFieldPrepender(4),
                            new RpcDecoder(Response.class), //
                            new RpcEncoder(Request.class), //
                            new IdleStateHandler(0, 0, 60, TimeUnit.SECONDS),
                            new NettyConnHandler(),
                            new NettyClientHandler());
                }
            });

    this.scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5, new ThreadFactory() {
        private final AtomicInteger idGenerator = new AtomicInteger(0);
        @Override
        public Thread newThread(Runnable r) {

            return new Thread(r, "Rpc-Scheduled-" + this.idGenerator.incrementAndGet());
        }
    });


    this.scheduledThreadPoolExecutor.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
            scanRpcFutureTable();
        }
    }, 500, 500, TimeUnit.MILLISECONDS);
}

@Override
protected void initChannel(SocketChannel channel) throws Exception {
    ChannelPipeline pipeline = channel.pipeline();
    pipeline.addLast("frame-decoder", new LengthFieldBasedFrameDecoder(200 * MB, 0, 4, 0, 4));
    pipeline.addLast("frame-encoder", new LengthFieldPrepender(4));
    pipeline.addLast(new RpcDecoder(RpcRequest.class)); // 解码 RPC 请求
    pipeline.addLast(new RpcEncoder(RpcResponse.class)); // 编码 RPC 响应
    // deal rpc request
    pipeline.addLast(new RpcServerHandler(handlerMap, workerExecutorService).setThreadPoolInfo(threadPoolInfo));
    // add heart ping,保证15s后先触发all_idle,在35s后就会触发IdleState.READER_IDLE（未读操作状态），此时服务器就会将通道关闭
    pipeline.addLast("ping-idle", new IdleStateHandler(15, 25, 35, TimeUnit.SECONDS));
    pipeline.addLast("heartBeat", new HeartBeatHandler());
}

protected void initChannel(Channel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();

    pipeline.addLast("frame-decoder", new LengthFieldBasedFrameDecoder(handlerConfig.getMaxFrameLengthMB() * MB, 0, 4, 0, 4));
    pipeline.addLast("frame-encoder", new LengthFieldPrepender(4));

       pipeline.addLast("pb-decoder", new RpcDecoder(RpcResponse.class)); // 解码 RPC 响应
       pipeline.addLast("pb-encoder", new RpcEncoder(RpcRequest.class)); // 编码 RPC 请求

    pipeline.addLast(RpcClientHandler.CLIENT_HANDLER_NAME, new RpcClientHandler(getResponsePromiseContainer()));
}