@Override
    public void invokeAsync(String addr, RemotingCommand request, long timeoutMillis, InvokeCallback invokeCallback)
        throws InterruptedException, RemotingConnectException, RemotingTooMuchRequestException, RemotingTimeoutException,
        RemotingSendRequestException {
        long beginStartTime = System.currentTimeMillis();
        final ChannelFuture channelFuture = this.getAndCreateChannelAsync(addr);
        if (channelFuture == null) {
            invokeCallback.operationFail(new RemotingConnectException(addr));
            return;
        }
        channelFuture.addListener(future -> {
            if (future.isSuccess()) {
                Channel channel = channelFuture.channel();
                String channelRemoteAddr = RemotingHelper.parseChannelRemoteAddr(channel);
                if (channel != null && channel.isActive()) {
                    long costTime = System.currentTimeMillis() - beginStartTime;
                    if (timeoutMillis < costTime) {
                        invokeCallback.operationFail(new RemotingTooMuchRequestException("invokeAsync call the addr[" + channelRemoteAddr + "] timeout"));
                    }
                    this.invokeAsyncImpl(channel, request, timeoutMillis - costTime, new InvokeCallbackWrapper(invokeCallback, addr));
                } else {
                    this.closeChannel(addr, channel);
                    invokeCallback.operationFail(new RemotingConnectException(addr));
                }
            } else {
                invokeCallback.operationFail(new RemotingConnectException(addr));
            }
        });
    }
@Test
    public void testInvokeAsync() {
        remotingClient.registerRPCHook(rpcHookMock);
        Channel channel = new LocalChannel();
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.PULL_MESSAGE, null);
        RemotingCommand response = RemotingCommand.createResponseCommand(null);
        response.setCode(ResponseCode.SUCCESS);
        ResponseFuture responseFuture = new ResponseFuture(channel, request.getOpaque(), request, 1000, new InvokeCallback() {
            @Override
            public void operationComplete(ResponseFuture responseFuture) {

            }
        }, new SemaphoreReleaseOnlyOnce(new Semaphore(1)));
        responseFuture.setResponseCommand(response);
        CompletableFuture<ResponseFuture> future = new CompletableFuture<>();
        future.complete(responseFuture);

        doReturn(future).when(remotingClient).invoke0(any(Channel.class), any(RemotingCommand.class), anyLong());

        InvokeCallback callback = mock(InvokeCallback.class);
        remotingClient.invokeAsyncImpl(channel, request, 1000, callback);
        verify(callback, times(1)).operationSucceed(eq(response));
        verify(callback, times(1)).operationComplete(eq(responseFuture));
        verify(callback, never()).operationFail(any());

        verify(rpcHookMock).doBeforeRequest(anyString(), eq(request));
        verify(rpcHookMock).doAfterResponse(anyString(), eq(request), eq(response));
    }