private Client createClient(final NodeAddrInfo addressInfo,
            int connectTimeout, final RpcConfig conf) throws Exception {
        final NettyClient client =
                new NettyClient(this, connectTimeout);
        Bootstrap clientBootstrap = new Bootstrap();
        clientBootstrap.group(eventLoopGroup);
        clientBootstrap.channel(EventLoopUtil.getClientSocketChannelClass(eventLoopGroup));
        clientBootstrap.option(ChannelOption.TCP_NODELAY, true);
        clientBootstrap.option(ChannelOption.SO_REUSEADDR, true);
        clientBootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, connectTimeout);

        int nettyWriteHighMark =
                conf.getInt(RpcConstants.NETTY_WRITE_HIGH_MARK, 64 * 1024);
        int nettyWriteLowMark =
                conf.getInt(RpcConstants.NETTY_WRITE_LOW_MARK, 32 * 1024);
        clientBootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK,
                new WriteBufferWaterMark(nettyWriteLowMark, nettyWriteHighMark));
        clientBootstrap.handler(new ChannelInitializer<SocketChannel>() {

            @Override
            public void initChannel(SocketChannel socketChannel) throws Exception {
                ChannelPipeline pipeline = socketChannel.pipeline();
                if (enableTLS) {
                    try {
                        SSLEngine sslEngine =
                                TSSLEngineUtil.createSSLEngine(keyStorePath, trustStorePath,
                                        keyStorePassword, trustStorePassword, true,
                                        needTwoWayAuthentic);
                        pipeline.addLast("ssl", new SslHandler(sslEngine));
                    } catch (Throwable t) {
                        logger.error(new StringBuilder(256)
                                .append("Create SSLEngine to connection ")
                                .append(addressInfo.getHostPortStr()).append(" failure!").toString(), t);
                        throw new Exception(t);
                    }
                }
                // Encode the data
                pipeline.addLast("protocolEncoder", new NettyProtocolEncoder());
                // Decode the bytes into a Rpc Data Pack
                pipeline.addLast("protocolDecoder", new NettyProtocolDecoder());
                // handle the time out requests
                pipeline.addLast("readTimeoutHandler", new ReadTimeoutHandler(
                        conf.getLong(RpcConstants.CONNECT_READ_IDLE_DURATION,
                                RpcConstants.CFG_CONNECT_READ_IDLE_TIME),
                        TimeUnit.MILLISECONDS));
                // tube netty client handler
                pipeline.addLast("clientHandler", client.new NettyClientHandler());
            }
        });
        ChannelFuture future =
                clientBootstrap.connect(new InetSocketAddress(addressInfo.getHost(), addressInfo.getPort()));
        future.awaitUninterruptibly(connectTimeout);
        if (!future.isDone()) {
            future.cancel(false);
            throw new LocalConnException(new StringBuilder(256).append("Create connection to ")
                    .append(addressInfo.getHostPortStr()).append(" timeout!").toString());
        }
        if (future.isCancelled()) {
            throw new LocalConnException(new StringBuilder(256).append("Create connection to ")
                    .append(addressInfo.getHostPortStr()).append(" cancelled by user!").toString());
        }
        if (!future.isSuccess()) {
            throw new LocalConnException(new StringBuilder(256).append("Create connection to ")
                    .append(addressInfo.getHostPortStr()).append(" error").toString(),
                    future.cause());
        }
        client.setChannel(future.channel(), addressInfo);
        return client;
    }
create a netty client

@param addressInfo        the remote address information
@param connectTimeout     the connection timeout
@param conf               the configure information
@return the client object
@throws Exception         the exception while creating object.
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClientFactory.java