private static <REQ> ByteBuf buildRequestBuf(String service, String version, String method, int seqid, REQ request, BeanSerializer<REQ> requestSerializer) throws SoaException {
    AbstractByteBufAllocator allocator =
            SoaSystemEnvProperties.SOA_POOLED_BYTEBUF ?
                    PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT;
    final ByteBuf requestBuf = allocator.buffer(8192);

    SoaMessageBuilder<REQ> builder = new SoaMessageBuilder<>();

    try {
        SoaHeader header = SoaHeaderHelper.buildHeader(service, version, method);

        ByteBuf buf = builder.buffer(requestBuf)
                .header(header)
                .body(request, requestSerializer)
                .seqid(seqid)
                .build();
        return buf;
    } catch (TException e) {
        e.printStackTrace();
    }

    return null;
}
 

protected Bootstrap initBootstrap() {
    AbstractByteBufAllocator allocator =
            SoaSystemEnvProperties.SOA_POOLED_BYTEBUF ?
                    PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT;
    bootstrap = new Bootstrap();
    bootstrap.group(workerGroup);
    bootstrap.channel(NioSocketChannel.class);
    bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
    bootstrap.option(ChannelOption.ALLOCATOR, allocator);
    bootstrap.handler(new ChannelInitializer<SocketChannel>() {
        @Override
        public void initChannel(SocketChannel ch) throws Exception {
            ch.pipeline().addLast(new IdleStateHandler(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds),
                    new SoaFrameDecoder(), //粘包和断包处理
                    new SoaIdleHandler(),
                    new SoaClientHandler(callBack));
        }
    });
    return bootstrap;
}
 

@Override
protected <REQ> ByteBuf buildRequestBuf(String service, String version, String method, int seqid, REQ request, BeanSerializer<REQ> requestSerializer) throws SoaException {
    AbstractByteBufAllocator allocator =
            SoaSystemEnvProperties.SOA_POOLED_BYTEBUF ?
                    PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT;
    final ByteBuf requestBuf = allocator.buffer(8192);

    SoaMessageBuilder<REQ> builder = new SoaMessageBuilder<>();

    try {
        SoaHeader header = SoaHeaderHelper.buildHeader(service, version, method);

        ByteBuf buf = builder.buffer(requestBuf)
                .header(header)
                .body(request, requestSerializer)
                .seqid(seqid)
                .build();
        return buf;
    } catch (TException e) {
        LOGGER.error(e.getMessage(), e);
        requestBuf.release();
        if (e instanceof SoaException) {
            throw (SoaException)e;
        } else {
            throw new SoaException(e);
        }
    }
}
 

@Override
public void start() {
    LOGGER.warn("Plugin::" + getClass().getSimpleName() + "::start");
    LOGGER.info("Bind Local Port {} [Netty]", port);
    LOGGER.info("ByteBufAllocator:{}", SoaSystemEnvProperties.SOA_POOLED_BYTEBUF ? "pooled" : "unpooled");

    Thread bootstrapThread = new Thread("NettyContainer-Thread") {
        @Override
        public void run() {
            try {
                bootstrap = new ServerBootstrap();

                AbstractByteBufAllocator allocator =
                        SoaSystemEnvProperties.SOA_POOLED_BYTEBUF ?
                                PooledByteBufAllocator.DEFAULT : UnpooledByteBufAllocator.DEFAULT;

                //链路控制
                ChannelHandler soaLinkStateHandler = new SoaLinkStateHandler();
                //编解码器
                ChannelHandler soaMsgDecoder = new SoaMsgDecoder(container);
                ChannelHandler soaMsgEncoder = new SoaMsgEncoder(container);

                //业务处理器
                ChannelHandler soaServerHandler = new SoaServerHandler(container);
                ChannelHandler soaInvokeCounter = MONITOR_ENABLE ? new SoaInvokeCounter() : null;

                //限流 handler
                SoaFreqHandler freqHandler = FREQ_LIMIT_ENABLE ? new SoaFreqHandler() : null;

                bootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                // 超时设置
                                ch.pipeline().addLast(HandlerConstants.IDLE_STATE_HANDLER, new IdleStateHandler(20, 0, 0));
                                //粘包和断包处理
                                ch.pipeline().addLast(HandlerConstants.SOA_FRAME_DECODER_HANDLER, new SoaFrameDecoder());
                                // 链路监控检测
                                ch.pipeline().addLast(HandlerConstants.SOA_IDLE_HANDLER, soaLinkStateHandler);
                                ch.pipeline().addLast(HandlerConstants.SOA_MSG_ENCODER_HANDLER, soaMsgEncoder);
                                ch.pipeline().addLast(HandlerConstants.SOA_MSG_DECODER_HANDLER, soaMsgDecoder);

                                if (FREQ_LIMIT_ENABLE) {
                                    // 添加服务限流handler
                                    ch.pipeline().addLast(HandlerConstants.SOA_FREQ_HANDLER, freqHandler);
                                }

                                // 服务调用统计
                                if (MONITOR_ENABLE) {
                                    ch.pipeline().addLast(HandlerConstants.SOA_INVOKE_COUNTER_HANDLER, soaInvokeCounter);
                                }

                                ch.pipeline().addLast(HandlerConstants.SOA_SERVER_HANDLER, soaServerHandler);
                            }
                        })
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .option(ChannelOption.ALLOCATOR, allocator)
                        .childOption(ChannelOption.SO_KEEPALIVE, true)
                        .childOption(ChannelOption.ALLOCATOR, allocator);

                // Start the server.
                ChannelFuture f = bootstrap.bind(port).sync();

                // Wait until the connection is closed.
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    };
    bootstrapThread.setDaemon(true);
    bootstrapThread.start();
}