class NettyRemotingServer extends NettyRemotingAbstract implements RemotingServer {

    private final NettyServerConfig nettyServerConfig;

    private final CommonThreadPool threadPool;

    private EventLoopGroup bossGroup;

    private EventLoopGroup workerGroup;

    private Channel channel = null;

    public NettyRemotingServer(final NettyServerConfig nettyServerConfig,
                               final NettyEventListener nettyEventListener,
                               final CommonThreadPool threadPool) {
        super(nettyEventListener);
        this.nettyServerConfig = nettyServerConfig;
        this.threadPool = threadPool;
    }

    @Override
    public void start() {

        this.threadPool.execute(() -> {
            int port = this.nettyServerConfig.getPort();
            ThreadFactory bossThreadFactory = new ThreadFactoryBuilder()
                    .setUncaughtExceptionHandler((thread, throwable) -> {
                        log.error("NettyServerBoss has uncaughtException.");
                        log.error(throwable.getMessage(), throwable);
                    })
                    .setDaemon(true)
                    .setNameFormat("netty-server-boss-%d")
                    .build();
            ThreadFactory workerThreadFactory = new ThreadFactoryBuilder()
                    .setUncaughtExceptionHandler((thread, throwable) -> {
                        log.error("NettyServerWorker has uncaughtException.");
                        log.error(throwable.getMessage(), throwable);
                    })
                    .setDaemon(true)
                    .setNameFormat("netty-server-worker-%d")
                    .build();
            if (this.useEpoll()) {
                bossGroup = new EpollEventLoopGroup(bossThreadFactory);
                workerGroup = new EpollEventLoopGroup(workerThreadFactory);
            } else {
                bossGroup = new NioEventLoopGroup(bossThreadFactory);
                workerGroup = new NioEventLoopGroup(workerThreadFactory);
            }

            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(this.useEpoll() ? EpollServerSocketChannel.class : NioServerSocketChannel.class)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childOption(ChannelOption.TCP_NODELAY, true)
                        .childOption(ChannelOption.SO_KEEPALIVE, false)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                NettyRemotingServer.this.initChannel(channel);
                            }
                        });
                channel = b.bind(port).sync().channel();
                channel.closeFuture().sync();
            } catch (InterruptedException ignored) {
                log.info("server shutdown now!");
            } catch (Exception e) {
                log.error("Netty Server start exception, {}", e.getMessage());
                throw new RuntimeException(e);
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        });
    }

    @Override
    public boolean isStart() {
        return this.channel != null && this.channel.isActive();
    }

    private void initChannel(final SocketChannel channel) {<FILL_FUNCTION_BODY>}

    @Override
    public void shutdown() {
        try {
            this.bossGroup.shutdownGracefully();

            this.workerGroup.shutdownGracefully();

            this.threadPool.destroy();

        } catch (Exception e) {
            log.error("Netty Server shutdown exception, ", e);
        }
    }

    @Override
    public void sendMsg(final Channel channel, final ClusterMsg.Message request) {
        this.sendMsgImpl(channel, request);
    }

    @Override
    public ClusterMsg.Message sendMsgSync(final Channel channel, final ClusterMsg.Message request, final int timeoutMillis) {
        return this.sendMsgSyncImpl(channel, request, timeoutMillis);
    }

    @Override
    public void registerHook(List<NettyHook> nettyHookList) {
        this.nettyHookList.addAll(nettyHookList);
    }

    /**
     * netty server handler
     */
    @ChannelHandler.Sharable
    public class NettyServerHandler extends SimpleChannelInboundHandler<ClusterMsg.Message> {

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            NettyRemotingServer.this.channelActive(ctx);
        }

        @Override
        protected void channelRead0(ChannelHandlerContext ctx, ClusterMsg.Message msg) throws Exception {
            NettyRemotingServer.this.processReceiveMsg(ctx, msg);
        }

        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
            NettyRemotingServer.this.channelIdle(ctx, evt);
        }
    }
}

        ChannelPipeline pipeline = channel.pipeline();
        // zip
        pipeline.addLast(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));
        pipeline.addLast(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));
        // protocol buf encode decode
        pipeline.addLast(new ProtobufVarint32FrameDecoder());
        pipeline.addLast(new ProtobufDecoder(ClusterMsg.Message.getDefaultInstance()));
        pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
        pipeline.addLast(new ProtobufEncoder());
        // idle state
        pipeline.addLast(new IdleStateHandler(0, 0, nettyServerConfig.getIdleStateEventTriggerTime()));
        pipeline.addLast(new NettyServerHandler());
    
<methods>public void registerProcessor(org.apache.hertzbeat.common.entity.message.ClusterMsg.MessageType, org.apache.hertzbeat.remoting.netty.NettyRemotingProcessor) <variables>protected org.apache.hertzbeat.remoting.event.NettyEventListener nettyEventListener,protected List<org.apache.hertzbeat.remoting.netty.NettyHook> nettyHookList,protected ConcurrentHashMap<org.apache.hertzbeat.common.entity.message.ClusterMsg.MessageType,org.apache.hertzbeat.remoting.netty.NettyRemotingProcessor> processorTable,protected ConcurrentHashMap<java.lang.String,org.apache.hertzbeat.remoting.netty.ResponseFuture> responseTable