@Override
    public ResponseWrapper call(RequestWrapper request, Callback callback,
            long timeout, TimeUnit timeUnit) throws Exception {
        if (closed.get()) {
            throw new ClientClosedException("Netty client has bean closed!");
        }
        request.setSerialNo(serialNoGenerator.incrementAndGet());
        RPCProtos.RpcConnHeader.Builder builder =
                RPCProtos.RpcConnHeader.newBuilder();
        builder.setFlag(request.getFlagId());
        final RPCProtos.RpcConnHeader connectionHeader =
                builder.build();
        RPCProtos.RequestHeader.Builder headerBuilder =
                RPCProtos.RequestHeader.newBuilder();
        headerBuilder.setServiceType(request.getServiceType());
        headerBuilder.setProtocolVer(request.getProtocolVersion());
        final RPCProtos.RequestHeader rpcHeader =
                headerBuilder.build();
        RPCProtos.RequestBody.Builder rpcBodyBuilder =
                RPCProtos.RequestBody.newBuilder();
        rpcBodyBuilder.setMethod(request.getMethodId());
        rpcBodyBuilder.setTimeout(request.getTimeout());
        rpcBodyBuilder
                .setRequest(ByteString.copyFrom(PbEnDecoder.pbEncode(request.getRequestData())));
        RPCProtos.RequestBody rpcBodyRequest = rpcBodyBuilder.build();
        ByteBufferOutputStream bbo = new ByteBufferOutputStream();
        connectionHeader.writeDelimitedTo(bbo);
        rpcHeader.writeDelimitedTo(bbo);
        rpcBodyRequest.writeDelimitedTo(bbo);
        RpcDataPack pack = new RpcDataPack(request.getSerialNo(), bbo.getBufferList());
        CallFuture<ResponseWrapper> future = new CallFuture<ResponseWrapper>(callback);
        requests.put(request.getSerialNo(), future);
        if (callback == null) {
            try {
                getChannel().writeAndFlush(pack);
                return future.get(timeout, timeUnit);
            } catch (Throwable e) {
                Callback<ResponseWrapper> callback1 =
                        requests.remove(request.getSerialNo());
                if (callback1 != null) {
                    if (closed.get()) {
                        throw new ClientClosedException("Netty client has bean closed!");
                    } else if (getChannel() == null) {
                        throw new ClientClosedException("Send failure for channel is null!");
                    } else {
                        throw e;
                    }
                }
            }
        } else {
            boolean inserted = false;
            try {
                timeouts.put(request.getSerialNo(),
                        timer.newTimeout(new TimeoutTask(request.getSerialNo()), timeout, timeUnit));
                inserted = true;
                // write data after build Timeout to avoid one request processed twice
                getChannel().writeAndFlush(pack);
            } catch (Throwable e) {
                Callback<ResponseWrapper> callback1 =
                        requests.remove(request.getSerialNo());
                if (callback1 != null) {
                    if (inserted) {
                        Timeout timeout1 = timeouts.remove(request.getSerialNo());
                        if (timeout1 != null) {
                            timeout1.cancel();
                        }
                    }
                    if (closed.get()) {
                        throw new ClientClosedException("Netty client has bean closed!");
                    } else if (getChannel() == null) {
                        throw new ClientClosedException("Channel is null!");
                    } else {
                        throw e;
                    }
                }
            }
        }
        return null;
    }
Set a channel

@param channel      the channel
@param addressInfo   the address of the channel
inlong-tubemq/tubemq-core/src/main/java/org/apache/inlong/tubemq/corerpc/netty/NettyClient.java