[网络协议]十一：Dubbo高级特性（一）异步无返回、异步、异步转同步、事件回调

异步无返回

配置

<dubbo:reference id="serviceDemo" interface="com.jiangzheng.course.dubbo.api.service.ServiceDemo">

    <dubbo:method name="sayHello" return="false"/>
</dubbo:reference>

注意此处有个坑：如果通过配置的dubbo:reference的话，在使用的时候 请勿在使用@DubboReference去引用接口实例，否则xml的配置不生效

源码解析

org.apache.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke

	@Override
    protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(PATH_KEY, getUrl().getPath());
        inv.setAttachment(VERSION_KEY, version);

        ExchangeClient currentClient;
        if (clients.length == 1) {
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
        	//isOneway为true 表示 接口无返回，isTwoway 表示请求有返回
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);//获取 RETURN_KEY： "return"的配置值
            int timeout = calculateTimeout(invocation, methodName);
            invocation.put(TIMEOUT_KEY, timeout);
            if (isOneway) {
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                //进行调用
                currentClient.send(inv, isSent);
                //我们主要看这里，进入方法
                return AsyncRpcResult.newDefaultAsyncResult(invocation);
            } else {
                ExecutorService executor = getCallbackExecutor(getUrl(), inv);
                CompletableFuture<AppResponse> appResponseFuture =
                        currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);
                // save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
                FutureContext.getContext().setCompatibleFuture(appResponseFuture);
                AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);
                result.setExecutor(executor);
                return result;
            }
        } catch (TimeoutException e) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (RemotingException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

    //org.apache.dubbo.rpc.AsyncRpcResult#newDefaultAsyncResult(org.apache.dubbo.rpc.Invocation)
    public static AsyncRpcResult newDefaultAsyncResult(Invocation invocation) {
    	//第一个参数 为返回的结果，此处设置为null， 第二个参数为 报错信息，也是设置为null，第三个是 consumer端的 invocation
        return newDefaultAsyncResult(null, null, invocation);
    }

    public static AsyncRpcResult newDefaultAsyncResult(Object value, Throwable t, Invocation invocation) {
        CompletableFuture<AppResponse> future = new CompletableFuture<>();
        AppResponse result = new AppResponse();
        if (t != null) {
            result.setException(t);
        } else {
        	//设置返回结果为null
            result.setValue(value);
        }
        future.complete(result);
        //在consumer端直接进行返回
        return new AsyncRpcResult(future, invocation);
    }

异步

配置

<dubbo:reference id="serviceDemo" interface="com.jiangzheng.course.dubbo.api.service.ServiceDemo">

    <dubbo:method name="getSelf" async="true"/>
</dubbo:reference>

调用

import com.jiangzheng.course.dubbo.api.service.ServiceDemo;
import org.apache.dubbo.rpc.RpcContext;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.ImportResource;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@SpringBootApplication
@ImportResource(locations = {"classpath:springContext-dubbo.xml"})
public class DubboDemoApplication {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ConfigurableApplicationContext context = SpringApplication.run(DubboDemoApplication.class, args);
        ServiceDemo serviceDemo = context.getBean("serviceDemo", ServiceDemo.class);
        //调用
        serviceDemo.getSelf("hello wwy");

        //获取异步结果
        CompletableFuture<String> completableFuture = RpcContext.getContext().getCompletableFuture();
        String result = completableFuture.get();
        System.out.println(result);
    }

}

源码解析

    //org.apache.dubbo.rpc.protocol.dubbo.DubboInvoker#doInvoke
    @Override
    protected Result doInvoke(final Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation) invocation;
        final String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment(PATH_KEY, getUrl().getPath());
        inv.setAttachment(VERSION_KEY, version);

        ExchangeClient currentClient;
        if (clients.length == 1) {
            currentClient = clients[0];
        } else {
            currentClient = clients[index.getAndIncrement() % clients.length];
        }
        try {
            boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
            int timeout = calculateTimeout(invocation, methodName);
            invocation.put(TIMEOUT_KEY, timeout);
            if (isOneway) {
                boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                currentClient.send(inv, isSent);
                return AsyncRpcResult.newDefaultAsyncResult(invocation);
            } else {//执行到此处
				
				//获取线程池
                ExecutorService executor = getCallbackExecutor(getUrl(), inv);
                //发起请求，我们进入此方法
                CompletableFuture<AppResponse> appResponseFuture =
                        currentClient.request(inv, timeout, executor).thenApply(obj -> (AppResponse) obj);
                // save for 2.6.x compatibility, for example, TraceFilter in Zipkin uses com.alibaba.xxx.FutureAdapter
                FutureContext.getContext().setCompatibleFuture(appResponseFuture);
                AsyncRpcResult result = new AsyncRpcResult(appResponseFuture, inv);
                result.setExecutor(executor);
                return result;
            }
        } catch (TimeoutException e) {
            throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (RemotingException e) {
            throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

    //org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeClient#request(java.lang.Object, int, java.util.concurrent.ExecutorService)
    @Override
    public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) throws RemotingException {
    	//继续调用
        return channel.request(request, timeout, executor);
    }

    //org.apache.dubbo.remoting.exchange.support.header.HeaderExchangeChannel#request(java.lang.Object, int, java.util.concurrent.ExecutorService)
    @Override
    public CompletableFuture<Object> request(Object request, int timeout, ExecutorService executor) throws RemotingException {
        if (closed) {
            throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " + request + ", cause: The channel " + this + " is closed!");
        }
        // 创建Request
        Request req = new Request();
        req.setVersion(Version.getProtocolVersion());
        req.setTwoWay(true);
        req.setData(request);
        //封装一个future，我们进入此方法
        DefaultFuture future = DefaultFuture.newFuture(channel, req, timeout, executor);
        try {
            channel.send(req);
        } catch (RemotingException e) {
            future.cancel();
            throw e;
        }
        return future;
    }

    //org.apache.dubbo.remoting.exchange.support.DefaultFuture#newFuture
    public static DefaultFuture newFuture(Channel channel, Request request, int timeout, ExecutorService executor) {
        //new 一个 future，我们看下future里面的属性信息
        final DefaultFuture future = new DefaultFuture(channel, request, timeout);
        future.setExecutor(executor);
        // ThreadlessExecutor needs to hold the waiting future in case of circuit return.
        if (executor instanceof ThreadlessExecutor) {
            ((ThreadlessExecutor) executor).setWaitingFuture(future);
        }
        // timeout check
        timeoutCheck(future);
        return future;
    }

    //org.apache.dubbo.remoting.exchange.support.DefaultFuture#DefaultFuture
    private DefaultFuture(Channel channel, Request request, int timeout) {
        this.channel = channel;
        this.request = request;
        this.id = request.getId();//请求id
        this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(TIMEOUT_KEY, DEFAULT_TIMEOUT);
        // put into waiting map.
        FUTURES.put(id, this);// 缓存 id->DefaultFuture
        CHANNELS.put(id, channel);// 缓存 id->channel
    }

我们直接看received方法，看下如何执行请求返回的

    //org.apache.dubbo.remoting.exchange.support.DefaultFuture#received(org.apache.dubbo.remoting.Channel, org.apache.dubbo.remoting.exchange.Response, boolean)
    public static void received(Channel channel, Response response, boolean timeout) {
        try {
        	//从缓存基于id获取 DefaultFuture（此处的response.getId() 于 上面提的 request.getId()是一致的）
            DefaultFuture future = FUTURES.remove(response.getId());
            if (future != null) {
                Timeout t = future.timeoutCheckTask;
                if (!timeout) {
                    // decrease Time
                    t.cancel();
                }
                //执行此处方法，我们进入查看
                future.doReceived(response);
            } else {
                logger.warn("The timeout response finally returned at "
                        + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
                        + ", response status is " + response.getStatus()
                        + (channel == null ? "" : ", channel: " + channel.getLocalAddress()
                        + " -> " + channel.getRemoteAddress()) + ", please check provider side for detailed result.");
            }
        } finally {
            CHANNELS.remove(response.getId());
        }
    }

    //org.apache.dubbo.remoting.exchange.support.DefaultFuture#doReceived
    private void doReceived(Response res) {
        if (res == null) {
            throw new IllegalStateException("response cannot be null");
        }
        //请求是否OK
        if (res.getStatus() == Response.OK) {
        	//唤醒并回填返回结果
            this.complete(res.getResult());
        } else if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
            this.completeExceptionally(new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage()));
        } else {
            this.completeExceptionally(new RemotingException(channel, res.getErrorMessage()));
        }

        // the result is returning, but the caller thread may still waiting
        // to avoid endless waiting for whatever reason, notify caller thread to return.
        if (executor != null && executor instanceof ThreadlessExecutor) {
            ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) executor;
            if (threadlessExecutor.isWaiting()) {
                threadlessExecutor.notifyReturn(new IllegalStateException("The result has returned, but the biz thread is still waiting" +
                        " which is not an expected state, interrupt the thread manually by returning an exception."));
            }
        }
    }

异步转同步

源码解析

针对dubbo的异步调用，但是是如何同步返回呢（即async="false"时），我们具体看下

在调用DubboInvoker前，会先调用AbstractInvoker的invoke

    //org.apache.dubbo.rpc.protocol.AbstractInvoker#invoke
    @Override
    public Result invoke(Invocation inv) throws RpcException {
        // if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
        if (isDestroyed()) {
            logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
                    + ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
        }
        RpcInvocation invocation = (RpcInvocation) inv;
        invocation.setInvoker(this);
        if (CollectionUtils.isNotEmptyMap(attachment)) {
            invocation.addObjectAttachmentsIfAbsent(attachment);
        }

        Map<String, Object> contextAttachments = RpcContext.getContext().getObjectAttachments();
        if (contextAttachments != null && contextAttachments.size() != 0) {
            invocation.addObjectAttachmentsIfAbsent(contextAttachments);
        }

        invocation.setInvokeMode(RpcUtils.getInvokeMode(url, invocation));
        RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);

        AsyncRpcResult asyncResult;
        try {
        	//调用DubboInvoker的doInvoke
            asyncResult = (AsyncRpcResult) doInvoke(invocation);
        } catch (InvocationTargetException e) { // biz exception
            Throwable te = e.getTargetException();
            if (te == null) {
                asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
            } else {
                if (te instanceof RpcException) {
                    ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
                }
                asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, te, invocation);
            }
        } catch (RpcException e) {
            if (e.isBiz()) {
                asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
            } else {
                throw e;
            }
        } catch (Throwable e) {
            asyncResult = AsyncRpcResult.newDefaultAsyncResult(null, e, invocation);
        }
        //设置返回的 Future
        RpcContext.getContext().setFuture(new FutureAdapter(asyncResult.getResponseFuture()));
		//同步的话，进行阻塞等待返回，我们进入此方法
        waitForResultIfSync(asyncResult, invocation);
        return asyncResult;
    }

    private void waitForResultIfSync(AsyncRpcResult asyncResult, RpcInvocation invocation) {
        try {
        	//如果请求模型为 true，即 同步的话
            if (InvokeMode.SYNC == invocation.getInvokeMode()) {
                /**
                 * NOTICE!
                 * must call {@link java.util.concurrent.CompletableFuture#get(long, TimeUnit)} because
                 * {@link java.util.concurrent.CompletableFuture#get()} was proved to have serious performance drop.
                 */
                 //阻塞获取结果
                asyncResult.get(Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
            }
        } catch (InterruptedException e) {
            throw new RpcException("Interrupted unexpectedly while waiting for remote result to return!  method: " +
                    invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
        } catch (ExecutionException e) {
            Throwable t = e.getCause();
            if (t instanceof TimeoutException) {
                throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " +
                        invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            } else if (t instanceof RemotingException) {
                throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " +
                        invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            } else {
                throw new RpcException(RpcException.UNKNOWN_EXCEPTION, "Fail to invoke remote method: " +
                        invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            }
        } catch (Throwable e) {
            throw new RpcException(e.getMessage(), e);
        }
    }

事件回调

配置

<dubbo:reference id="serviceDemo" interface="com.jiangzheng.course.dubbo.api.service.ServiceDemo">

    <dubbo:method name="getSelf" async="true"
                  onreturn="providerCallbackService.onReturn"
                  onthrow="providerCallbackService.onThrow" 
                  oninvoke="providerCallbackService.onInvoke"/>
</dubbo:reference>

import org.springframework.stereotype.Service;

@Service("providerCallbackService")
public class ProviderCallbackServiceImpl {

    //可以有多个参数，但是第一个参数一定要与实际调用方法的返回值一致，多个参数则会填充实际调用方法的入参
    public void onReturn(String response) {
        System.out.println("onReturn : " + response);
    }
    
    //可以有多个参数，但是第一个参数一定是Throwable，多个参数则会填充实际调用方法的入参
    public void onThrow(Throwable throwable) {
        System.out.println("onThrow : " + throwable.getMessage());
    }

	//参数类型与实际的调用方法要求一致
    public void onInvoke(String message) {
        System.out.println("onInvoke : " + message);
    }

}

@SpringBootApplication
@ImportResource(locations = {"classpath:springContext-dubbo.xml"})
public class DubboDemoApplication {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        ConfigurableApplicationContext context = SpringApplication.run(DubboDemoApplication.class, args);
        ServiceDemo serviceDemo = context.getBean("serviceDemo", ServiceDemo.class);
        //调用
        String result = serviceDemo.getSelf("hello wwy");
        System.out.println(result);
    }

}

onInvoke : hello wwy
onReturn : hello wwy

源码解析

此处实现在org.apache.dubbo.rpc.protocol.dubbo.filter.FutureFilter中

//@Activate说明是默认激活的状态，CommonConstants.CONSUMER说明此filter用于调用端
@Activate(group = CommonConstants.CONSUMER)
public class FutureFilter implements ClusterFilter, ClusterFilter.Listener {

	@Override //对应了 onInvoke
    public Result invoke(final Invoker<?> invoker, final Invocation invocation) throws RpcException {
        //1、主要看这个方法，我们进入此方法
        fireInvokeCallback(invoker, invocation);
        //3、继续向下调用
        return invoker.invoke(invocation);
    }
    
    @Override //对应了 onReturn
    public void onResponse(Result result, Invoker<?> invoker, Invocation invocation) {
        //4、
        //判断是否存在异常
        if (result.hasException()) {
        	//5、进入此方法
            fireThrowCallback(invoker, invocation, result.getException());
        } else {
        	//8、进入此方法
            fireReturnCallback(invoker, invocation, result.getValue());
        }
    }

    @Override //对应了 onThrow
    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
    	//6、或者进入此方法
    	fireThrowCallback(invoker, invocation, t);
    }
	
	private void fireInvokeCallback(final Invoker<?> invoker, final Invocation invocation) {
		//2
		//ASYNC_METHOD_INFO = "async-method-info"
	    AsyncMethodInfo asyncMethodInfo = (AsyncMethodInfo) invocation.get(ASYNC_METHOD_INFO);
        if (asyncMethodInfo != null) {
            return asyncMethodInfo;
        }
		//获取ConsumerModel，
		//ConsumerModel中包含属性：Map<String, AsyncMethodInfo> methodConfigs，方法集合
		//AsyncMethodInfo包含oninvokeMethod、onreturnMethod、onthrowMethod，其中 这些属性对应的是就是 xml中配置的方法
        ConsumerModel consumerModel = ApplicationModel.getConsumerModel(invoker.getUrl().getServiceKey());
        if (consumerModel == null) {
            return null;
        }

        String methodName = invocation.getMethodName();
        if (methodName.equals($INVOKE)) {
            methodName = (String) invocation.getArguments()[0];
        }

        //获取到AsyncMethodInfo的实例对象
        final AsyncMethodInfo asyncMethodInfo = consumerModel.getAsyncInfo(methodName);
        
        if (asyncMethodInfo == null) {
            return;
        }
        //获取onInvokeMethod即回到方法、onInvokeInst即回调的方法所在的实例对象
        final Method onInvokeMethod = asyncMethodInfo.getOninvokeMethod();
        final Object onInvokeInst = asyncMethodInfo.getOninvokeInstance();

        if (onInvokeMethod == null && onInvokeInst == null) {
            return;
        }
        if (onInvokeMethod == null || onInvokeInst == null) {
            throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a oninvoke callback config , but no such " + (onInvokeMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
        }
        if (!onInvokeMethod.isAccessible()) {
            onInvokeMethod.setAccessible(true);
        }
		//获取请求参数
        Object[] params = invocation.getArguments();
        try {
        	//执行onInvoke的回调方法
            onInvokeMethod.invoke(onInvokeInst, params);
        } catch (InvocationTargetException e) {
            fireThrowCallback(invoker, invocation, e.getTargetException());
        } catch (Throwable e) {
            fireThrowCallback(invoker, invocation, e);
        }
    }

    private void fireThrowCallback(final Invoker<?> invoker, final Invocation invocation, final Throwable exception) {
        //7、
        //获取配置的 onthrow方法信息
        final AsyncMethodInfo asyncMethodInfo = getAsyncMethodInfo(invoker, invocation);
        if (asyncMethodInfo == null) {
            return;
        }

        final Method onthrowMethod = asyncMethodInfo.getOnthrowMethod();
        final Object onthrowInst = asyncMethodInfo.getOnthrowInstance();

        //onthrow callback not configured
        if (onthrowMethod == null && onthrowInst == null) {
            return;
        }
        if (onthrowMethod == null || onthrowInst == null) {
            throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onthrow callback config , but no such " + (onthrowMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
        }
        if (!onthrowMethod.isAccessible()) {
            onthrowMethod.setAccessible(true);
        }
        Class<?>[] rParaTypes = onthrowMethod.getParameterTypes();
        //第一个的参数类型必须是exception类型的
        if (rParaTypes[0].isAssignableFrom(exception.getClass())) {
            try {
                Object[] args = invocation.getArguments();
                Object[] params;

				//如果onthrow方法入参长度是大于1的话
                if (rParaTypes.length > 1) {
                	//如果onthrow方法入参长度是等于2的话，并且 第二个参数是 数组类型，则第一个参数 设置为exception，第二个为方法调用入参
                    if (rParaTypes.length == 2 && rParaTypes[1].isAssignableFrom(Object[].class)) {
                        params = new Object[2];
                        params[0] = exception;
                        params[1] = args;
                    } else {
                    	//否则 在方法调用参数基础上加一，将exception加入，并放在第一位
                        params = new Object[args.length + 1];
                        params[0] = exception;
                        System.arraycopy(args, 0, params, 1, args.length);
                    }
                } else {
                	//如果onthrow方法入参长度是等于1的话，直接将exception传入
                    params = new Object[]{exception};
                }
                //执行调用
                onthrowMethod.invoke(onthrowInst, params);
            } catch (Throwable e) {
                logger.error(invocation.getMethodName() + ".call back method invoke error . callback method :" + onthrowMethod + ", url:" + invoker.getUrl(), e);
            }
        } else {
            logger.error(invocation.getMethodName() + ".call back method invoke error . callback method :" + onthrowMethod + ", url:" + invoker.getUrl(), exception);
        }
    }
    
    private void fireReturnCallback(final Invoker<?> invoker, final Invocation invocation, final Object result) {
        //8、
        //此处与 onthrow类似，不再赘述
        final AsyncMethodInfo asyncMethodInfo = getAsyncMethodInfo(invoker, invocation);
        if (asyncMethodInfo == null) {
            return;
        }

        final Method onReturnMethod = asyncMethodInfo.getOnreturnMethod();
        final Object onReturnInst = asyncMethodInfo.getOnreturnInstance();

        //not set onreturn callback
        if (onReturnMethod == null && onReturnInst == null) {
            return;
        }

        if (onReturnMethod == null || onReturnInst == null) {
            throw new IllegalStateException("service:" + invoker.getUrl().getServiceKey() + " has a onreturn callback config , but no such " + (onReturnMethod == null ? "method" : "instance") + " found. url:" + invoker.getUrl());
        }
        if (!onReturnMethod.isAccessible()) {
            onReturnMethod.setAccessible(true);
        }

        Object[] args = invocation.getArguments();
        Object[] params;
        Class<?>[] rParaTypes = onReturnMethod.getParameterTypes();
        if (rParaTypes.length > 1) {
            if (rParaTypes.length == 2 && rParaTypes[1].isAssignableFrom(Object[].class)) {
                params = new Object[2];
                params[0] = result;
                params[1] = args;
            } else {
                params = new Object[args.length + 1];
                params[0] = result;
                System.arraycopy(args, 0, params, 1, args.length);
            }
        } else {
            params = new Object[]{result};
        }
        try {
            onReturnMethod.invoke(onReturnInst, params);
        } catch (InvocationTargetException e) {
            fireThrowCallback(invoker, invocation, e.getTargetException());
        } catch (Throwable e) {
            fireThrowCallback(invoker, invocation, e);
        }
    }
}

异步编程的一些其他写法示例可以查看官网：https://dubbo.apache.org/zh/docs/advanced/async-call/