class NettyRemotingClient extends NettyRemotingAbstract implements RemotingClient {

    private final NettyClientConfig nettyClientConfig;

    private final CommonThreadPool threadPool;

    private final Bootstrap bootstrap = new Bootstrap();

    private EventLoopGroup workerGroup;

    private Channel channel;

    public NettyRemotingClient(final NettyClientConfig nettyClientConfig,
                               final NettyEventListener nettyEventListener,
                               final CommonThreadPool threadPool) {
        super(nettyEventListener);
        this.nettyClientConfig = nettyClientConfig;
        this.threadPool = threadPool;
    }

    @Override
    public void start() {<FILL_FUNCTION_BODY>}

    private void initChannel(final SocketChannel channel) {
        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());
        pipeline.addLast(new NettyClientHandler());

    }

    @Override
    public void shutdown() {
        try {
            if (this.channel != null) {
                this.channel.close();
            }

            this.workerGroup.shutdownGracefully();

            this.threadPool.destroy();

        } catch (Exception e) {
            log.error("netty client shutdown exception, ", e);
        }
    }

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

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

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

    class NettyClientHandler extends SimpleChannelInboundHandler<ClusterMsg.Message> {

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

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

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

        this.threadPool.execute(() -> {
            ThreadFactory threadFactory = new ThreadFactoryBuilder()
                    .setUncaughtExceptionHandler((thread, throwable) -> {
                        log.error("NettyClientWorker has uncaughtException.");
                        log.error(throwable.getMessage(), throwable);
                    })
                    .setDaemon(true)
                    .setNameFormat("netty-client-worker-%d")
                    .build();
            this.workerGroup = new NioEventLoopGroup(threadFactory);
            this.bootstrap.group(workerGroup)
                    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, this.nettyClientConfig.getConnectTimeoutMillis())
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel channel) throws Exception {
                            NettyRemotingClient.this.initChannel(channel);
                        }
                    });

            this.channel = null;
            boolean first = true;
            while (!Thread.currentThread().isInterrupted() 
                    && (first || this.channel == null || !this.channel.isActive())) {
                first = false;
                try {
                    this.channel = this.bootstrap
                            .connect(this.nettyClientConfig.getServerHost(), this.nettyClientConfig.getServerPort())
                            .sync().channel();
                    this.channel.closeFuture().sync();
                } catch (InterruptedException ignored) {
                    log.info("client shutdown now!");
                    Thread.currentThread().interrupt();
                } catch (Exception e2) {
                    log.error("client connect to server error: {}. try after 10s.", e2.getMessage());
                    try {
                        Thread.sleep(10000);
                    } catch (InterruptedException ignored) {
                    }
                }
            }
            workerGroup.shutdownGracefully();
        });
    
<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