DUBBO消费异步化解决了什么问题
欢迎大家关注公众号「JAVA前线」查看更多精彩分享文章，主要包括源码分析、实际应用、架构思维、职场分享、产品思考等等，同时欢迎大家加我个人微信「java_front」一起交流学习 1 文章概述 我们在服务端开发时如果需要实现异步调用，首先声明一个线程池，并将调用业务方法封装成一个任务提交至线程池，如果不需要获取返回值则封装为Runnable，需要获取返回值则封装为Callable并通过Future对象接受结果。 classCalcTask1implementsCallable<Integer>{ @Override publicIntegercall()throwsException{ System.out.println("task1耗时计算"); Thread.sleep(1000L); return100; } } classCalcTask2implementsCallable<Integer>{ @Override publicIntegercall()throwsException{ System.out.println("task2耗时计算"); Thread.sleep(3000L); return200; } } publicclassCallableTest{ publicstaticvoidtest1()throwsException{ ExecutorServiceexecutor=Executors.newCachedThreadPool(); CalcTask1task1=newCalcTask1(); Future<Integer>f1=executor.submit(task1); CalcTask2task2=newCalcTask2(); Future<Integer>f2=executor.submit(task2); Integerresult1=f1.get(); Integerresult2=f2.get(); System.out.println("finalresult="+(result1+result2)); executor.shutdown(); } publicstaticvoidtest2()throwsException{ ExecutorServiceexecutor=Executors.newCachedThreadPool(); List<Callable<Integer>>tasks=newArrayList<Callable<Integer>>(); CalcTask1task1=newCalcTask1(); CalcTask2task2=newCalcTask2(); tasks.add(task1); tasks.add(task2); for(inti=0;i<tasks.size();i++){ Future<Integer>future=executor.submit(tasks.get(i)); System.out.println("result="+future.get()); } executor.shutdown(); } } 1.1 什么是消费异步化 在使用DUBBO进行异步化调用时不需要这么麻烦，DUBBO基于NIO非阻塞能力使得服务消费者无需启用多线程就可以实现并行调用多个服务，在此我们给出基于2.7.0版本调用实例。 1.1.1 生产者 (1) 服务声明 publicinterfaceCalcSumService{ publicIntegersum(inta,intb); } publicclassCalcSumServiceImplimplementsCalcSumService{ @Override publicIntegersum(inta,intb){ returna+b; } } publicinterfaceCalcSubtractionService{ publicIntegersubtraction(inta,intb); } publicclassCalcSubtractionServiceImplimplementsCalcSubtractionService{ @Override publicIntegersubtraction(inta,intb){ returna-b; } } (2) 配置文件 <beans> <dubbo:applicationname="java-front-provider"/> <dubbo:registryaddress="zookeeper://127.0.0.1:2181"/> <dubbo:protocolname="dubbo"port="9999"/> <beanid="calcSumService"class="com.java.front.dubbo.demo.provider.service.CalcSumServiceImpl"/> <beanid="calcSubtractionService"class="com.java.front.dubbo.demo.provider.service.CalcSubtractionServiceImpl"/> <dubbo:serviceinterface="com.java.front.dubbo.demo.provider.service.CalcSumService"ref="calcSumService"/> <dubbo:serviceinterface="com.java.front.dubbo.demo.provider.service.CalcSubtractionService"ref="calcSubtractionService"/> </beans> (3) 服务发布 publicclassProvider{ publicstaticvoidmain(String[]args)throwsException{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext("classpath*:META-INF/spring/dubbo-provider.xml"); context.start(); System.out.println(context); System.in.read(); } } 1.1.2 消费者 (1) 配置文件 <beans> <dubbo:applicationname="java-front-consumer"/> <dubbo:registryaddress="zookeeper://127.0.0.1:2181"/> <dubbo:referenceid="calcSumService"interface="com.java.front.dubbo.demo.provider.service.CalcSumService"timeout="10000"> <dubbo:methodname="sum"async="true"/> </dubbo:reference> <dubbo:referenceid="calcSubtractionService"interface="com.java.front.dubbo.demo.provider.service.CalcSubtractionService"timeout="10000"> <dubbo:methodname="subtraction"async="true"/> </dubbo:reference> </beans> (2) 服务消费 publicclassConsumer{ publicstaticvoidmain(String[]args)throwsException{ testAsync(); System.in.read(); } publicstaticvoidtestAsync(){ try{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext(newString[]{"classpath*:META-INF/spring/dubbo-consumer.xml"}); System.out.println(context); context.start(); /**加法运算**/ CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); CompletableFuture<Integer>futureSum=RpcContext.getContext().getCompletableFuture(); /**减法运算**/ CalcSubtractionServicecalcSubtractionService=(CalcSubtractionService)context.getBean("calcSubtractionService"); calcSubtractionService.subtraction(3,2); CompletableFuture<Integer>futureSubtraction=RpcContext.getContext().getCompletableFuture(); /**输出结果**/ intsumResult=futureSum.get(); intsubtractionResult=futureSubtraction.get(); System.out.println("sumResult="+sumResult+",subtractionResult="+subtractionResult); }catch(Exceptione){ e.printStackTrace(); } } } 1.2 为什么消费异步化 异步化可以将原本串行的调用并行化，减少执行时间从而提升性能。假设上述实例加法服务需要100ms，减法服务需要200ms，那么串行化执行时间为二者之和300ms： 如果消费异步化那么执行时间减少为二者最大值200ms，异步化所带来的性能提升不言而喻： 2 保护性暂停模式 分析DUBBO源码之前我们首先介绍一种多线程设计模式：保护性暂停模式。我们设想这样一种场景：线程A生产数据，线程B读取这个数据。我们必须面对一种情况：线程B准备读取数据时，此时线程A还没有生产出数据。在这种情况下线程B不能一直空转，也不能立即退出，线程B要等到生产数据完成并拿到数据之后才退出。 那么在数据没有生产出这段时间，线程B需要执行一种等待机制，这样可以达到对系统保护目的，这就是保护性暂停。 publicclassMyDataimplementsSerializable{ privatestaticfinallongserialVersionUID=1L; privateStringmessage; publicMyData(Stringmessage){ this.message=message; } } classResource{ privateMyDatadata; privateObjectlock=newObject(); publicMyDatagetData(){ synchronized(lock){ while(data==null){ try{ //没有数据则释放锁并暂停等待被唤醒 lock.wait(); }catch(InterruptedExceptione){ e.printStackTrace(); } } returndata; } } publicvoidsendData(MyDatadata){ synchronized(lock){ //生产数据后唤醒消费线程 this.data=data; lock.notifyAll(); } } } publicclassProtectDesignTest{ publicstaticvoidmain(String[]args){ Resourceresource=newResource(); newThread(()->{ try{ MyDatadata=newMyData("hello"); System.out.println(Thread.currentThread().getName()+"生产数据="+data); //模拟发送耗时 TimeUnit.SECONDS.sleep(3); resource.sendData(data); }catch(InterruptedExceptione){ e.printStackTrace(); } },"t1").start(); newThread(()->{ MyDatadata=resource.getData(); System.out.println(Thread.currentThread().getName()+"接收到数据="+data); },"t2").start(); } } 在上述代码实例中线程1生产数据，线程2消费数据，Resource类通过wait/notify实现了保护性暂停模式，关于保护性暂停模式请参看我之前《保护性暂停模式详解以及其在DUBBO应用源码分析》这篇文章。 3 源码分析 本章节我们分析对比2.6.9和2.7.0两个版本源码，之所以选取这两个版本是因为2.7.0是一个里程碑版本，异步化能力得到了明显增强。 3.1 version_2.6.9 3.1.1 异步调用 我们首先看看这个版本异步调用使用方式，生产者内容和消费者配置文件同第一章节不再赘述，我们重点分析服务消费代码。 publicclassAsyncConsumer{ publicstaticvoidmain(String[]args)throwsException{ test1(); System.in.read(); } publicstaticvoidtest1()throwsException{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext(newString[]{"classpath*:META-INF/spring/dubbo-consumer.xml"}); System.out.println(context); context.start(); /**加法运算**/ CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); Future<Integer>futureSum=RpcContext.getContext().getFuture(); /**减法运算**/ CalcSubtractionServicecalcSubtractionService=(CalcSubtractionService)context.getBean("calcSubtractionService"); calcSubtractionService.subtraction(3,2); Future<Integer>futureSubtraction=RpcContext.getContext().getFuture(); /**输出结果**/ intsumResult=futureSum.get(); intsubtractionResult=futureSubtraction.get(); System.out.println("sumResult="+sumResult+",subtractionResult="+subtractionResult); } } 消费者最终执行DubboInvoker.doInvoke，这个方法包含异步调用核心： publicclassDubboInvoker<T>extendsAbstractInvoker<T>{ @Override protectedResultdoInvoke(finalInvocationinvocation)throwsThrowable{ RpcInvocationinv=(RpcInvocation)invocation; finalStringmethodName=RpcUtils.getMethodName(invocation); inv.setAttachment(Constants.PATH_KEY,getUrl().getPath()); inv.setAttachment(Constants.VERSION_KEY,version); ExchangeClientcurrentClient; if(clients.length==1){ currentClient=clients[0]; }else{ currentClient=clients[index.getAndIncrement()%clients.length]; } try{ booleanisAsync=RpcUtils.isAsync(getUrl(),invocation); booleanisOneway=RpcUtils.isOneway(getUrl(),invocation); inttimeout=getUrl().getMethodParameter(methodName,Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT); //单向调用 if(isOneway){ booleanisSent=getUrl().getMethodParameter(methodName,Constants.SENT_KEY,false); currentClient.send(inv,isSent); RpcContext.getContext().setFuture(null); returnnewRpcResult(); } //异步调用 elseif(isAsync){ //发起请求给生产者 ResponseFuturefuture=currentClient.request(inv,timeout); //设置future对象至上下文 RpcContext.getContext().setFuture(newFutureAdapter<Object>(future)); //返回空结果 returnnewRpcResult(); } //同步调用 else{ RpcContext.getContext().setFuture(null); return(Result)currentClient.request(inv,timeout).get(); } }catch(TimeoutExceptione){ thrownewRpcException(RpcException.TIMEOUT_EXCEPTION,"Invokeremotemethodtimeout.method:"+invocation.getMethodName()+",provider:"+getUrl()+",cause:"+e.getMessage(),e); }catch(RemotingExceptione){ thrownewRpcException(RpcException.NETWORK_EXCEPTION,"Failedtoinvokeremotemethod:"+invocation.getMethodName()+",provider:"+getUrl()+",cause:"+e.getMessage(),e); } } } 如果包含async属性则表示异步调用，第一步发送调用请求给生产者，第二步设置Future对象至上下文，第三步立即返回空结果。那么在服务消费时关键一步就是获取Future对象，所以我们在调用时要从上下文获取Future对象： CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); Future<Integer>futureSum=RpcContext.getContext().getFuture(); 使用Future对象获取结果： intsumResult=futureSum.get(); 进入FutureAdapter.get()方法： publicclassFutureAdapter<V>implementsFuture<V>{ privatefinalResponseFuturefuture; publicVget()throwsInterruptedException,ExecutionException{ try{ return(V)(((Result)future.get()).recreate()); }catch(RemotingExceptione){ thrownewExecutionException(e.getMessage(),e); }catch(Throwablee){ thrownewRpcException(e); } } } 进入ResponseFuture.get()方法，我们可以看到保护性暂停模式应用，当生产者线程没有返回数据则阻塞并等待被唤醒： publicclassDefaultFutureimplementsResponseFuture{ privatefinalLocklock=newReentrantLock(); privatefinalConditiondone=lock.newCondition(); @Override publicObjectget()throwsRemotingException{ returnget(timeout); } @Override publicObjectget(inttimeout)throwsRemotingException{ if(timeout<=0){ timeout=Constants.DEFAULT_TIMEOUT; } if(!isDone()){ longstart=System.currentTimeMillis(); lock.lock(); try{ while(!isDone()){ //远程调用未完成则等待被唤醒 done.await(timeout,TimeUnit.MILLISECONDS); //超时时间未完成则退出 if(isDone()||System.currentTimeMillis()-start>timeout){ break; } } }catch(InterruptedExceptione){ thrownewRuntimeException(e); }finally{ lock.unlock(); } //抛出超时异常 if(!isDone()){ thrownewTimeoutException(sent>0,channel,getTimeoutMessage(false)); } } returnreturnFromResponse(); } } 当消费者接收到生产者响应时会调用received方法唤醒相关阻塞线程，这时阻塞在get方法中的线程即可获取到数据： publicclassDefaultFutureimplementsResponseFuture{ privatefinalLocklock=newReentrantLock(); privatefinalConditiondone=lock.newCondition(); publicstaticvoidreceived(Channelchannel,Responseresponse){ try{ //根据唯一请求号获取Future DefaultFuturefuture=FUTURES.remove(response.getId()); if(future!=null){ future.doReceived(response); }else{ logger.warn("Thetimeoutresponsefinallyreturnedat" +(newSimpleDateFormat("yyyy-MM-ddHH:mm:ss.SSS").format(newDate())) +",response"+response +(channel==null?"":",channel:"+channel.getLocalAddress() +"->"+channel.getRemoteAddress())); } }finally{ CHANNELS.remove(response.getId()); } } privatevoiddoReceived(Responseres){ lock.lock(); try{ response=res; if(done!=null){ //唤醒相关阻塞线程 done.signal(); } }finally{ lock.unlock(); } if(callback!=null){ invokeCallback(callback); } } } 3.1.2 设置回调函数 我们现在调用get方法会阻塞在那里等到结果，那么有没有一种方式当结果返回时就立即调用我们设置的回调函数？答案是有。 publicclassAsyncConsumer{ publicstaticvoidmain(String[]args)throwsException{ test2(); System.in.read(); } publicstaticvoidtest2()throwsException{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext(newString[]{"classpath*:META-INF/spring/dubbo-consumer.xml"}); System.out.println(context); context.start(); /**加法运算**/ CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); /**执行回调函数**/ ((FutureAdapter<Object>)RpcContext.getContext().getFuture()).getFuture().setCallback(newResponseCallback(){ @Override publicvoiddone(Objectresponse){ System.out.println("sumResult="+response); } @Override publicvoidcaught(Throwableexception){ exception.printStackTrace(); } }); /**减法运算**/ CalcSubtractionServicecalcSubtractionService=(CalcSubtractionService)context.getBean("calcSubtractionService"); calcSubtractionService.subtraction(3,2); /**执行回调函数**/ ((FutureAdapter<Object>)RpcContext.getContext().getFuture()).getFuture().setCallback(newResponseCallback(){ @Override publicvoiddone(Objectresponse){ System.out.println("subtractionResult="+response); } @Override publicvoidcaught(Throwableexception){ exception.printStackTrace(); } }); } } DefaultFuture可以设置callback回调函数，当结果返回时如果回调函数不为空则执行： publicclassDefaultFutureimplementsResponseFuture{ privatevolatileResponseCallbackcallback; privatevoiddoReceived(Responseres){ lock.lock(); try{ response=res; if(done!=null){ done.signal(); } }finally{ lock.unlock(); } if(callback!=null){ //执行回调函数 invokeCallback(callback); } } privatevoidinvokeCallback(ResponseCallbackc){ ResponseCallbackcallbackCopy=c; if(callbackCopy==null){ thrownewNullPointerException("callbackcannotbenull."); } c=null; Responseres=response; if(res==null){ thrownewIllegalStateException("responsecannotbenull.url:"+channel.getUrl()); } if(res.getStatus()==Response.OK){ try{ //执行成功回调 callbackCopy.done(res.getResult()); }catch(Exceptione){ logger.error("callbackinvokeerror.reasult:"+res.getResult()+",url:"+channel.getUrl(),e); } }elseif(res.getStatus()==Response.CLIENT_TIMEOUT||res.getStatus()==Response.SERVER_TIMEOUT){ try{ TimeoutExceptionte=newTimeoutException(res.getStatus()==Response.SERVER_TIMEOUT,channel,res.getErrorMessage()); //发生超时回调 callbackCopy.caught(te); }catch(Exceptione){ logger.error("callbackinvokeerror,url:"+channel.getUrl(),e); } }else{ try{ RuntimeExceptionre=newRuntimeException(res.getErrorMessage()); callbackCopy.caught(re); }catch(Exceptione){ logger.error("callbackinvokeerror,url:"+channel.getUrl(),e); } } } } 3.2 version_2.7.0 CompletableFuture在这个版本中被引入实现异步调用，可以使用此类强大的异步编程API增强异步能力，我们首先回顾1.1.2章节实例： publicclassConsumer{ publicstaticvoidtestAsync(){ try{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext(newString[]{"classpath*:META-INF/spring/dubbo-consumer.xml"}); System.out.println(context); context.start(); /**加法运算**/ CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); CompletableFuture<Integer>futureSum=RpcContext.getContext().getCompletableFuture(); /**减法运算**/ CalcSubtractionServicecalcSubtractionService=(CalcSubtractionService)context.getBean("calcSubtractionService"); calcSubtractionService.subtraction(3,2); CompletableFuture<Integer>futureSubtraction=RpcContext.getContext().getCompletableFuture(); /**输出结果**/ intsumResult=futureSum.get(); intsubtractionResult=futureSubtraction.get(); System.out.println("sumResult="+sumResult+",subtractionResult="+subtractionResult); }catch(Exceptione){ e.printStackTrace(); } } } 在上述消费者代码的实例中我们只是应用了CompletableFuture.get()方法，并没有发挥其强大功能。我们对上述实例稍加改造，两个CompletionStage任务都执行完成后，两个任务结果会一起交给thenCombine进行处理： publicclassConsumer{ publicstaticvoidtestAsync(){ try{ ClassPathXmlApplicationContextcontext=newClassPathXmlApplicationContext(newString[]{"classpath*:META-INF/spring/dubbo-consumer2.xml"}); System.out.println(context); context.start(); /**加法运算**/ CalcSumServicecalcSumService=(CalcSumService)context.getBean("calcSumService"); calcSumService.sum(3,2); CompletableFuture<Integer>futureSum=RpcContext.getContext().getCompletableFuture(); /**减法运算**/ CalcSubtractionServicecalcSubtractionService=(CalcSubtractionService)context.getBean("calcSubtractionService"); calcSubtractionService.subtraction(3,2); CompletableFuture<Integer>futureSubtraction=RpcContext.getContext().getCompletableFuture(); /**乘法运算**/ CompletableFuture<Integer>multiplyResult=futureSum.thenCombine(futureSubtraction,newBiFunction<Integer,Integer,Integer>(){ @Override publicIntegerapply(Integert,Integeru){ return(t*u); } }); System.out.println("multiplyResult="+multiplyResult.get()); }catch(Exceptione){ e.printStackTrace(); } } } DubboInvoker代码有所变化： publicclassDubboInvoker<T>extendsAbstractInvoker<T>{ @Override protectedResultdoInvoke(finalInvocationinvocation)throwsThrowable{ RpcInvocationinv=(RpcInvocation)invocation; finalStringmethodName=RpcUtils.getMethodName(invocation); inv.setAttachment(Constants.PATH_KEY,getUrl().getPath()); inv.setAttachment(Constants.VERSION_KEY,version); ExchangeClientcurrentClient; if(clients.length==1){ currentClient=clients[0]; }else{ currentClient=clients[index.getAndIncrement()%clients.length]; } try{ //是否为异步调用 booleanisAsync=RpcUtils.isAsync(getUrl(),invocation); //是否为future异步方式 booleanisAsyncFuture=RpcUtils.isReturnTypeFuture(inv); //是否需要响应结果 booleanisOneway=RpcUtils.isOneway(getUrl(),invocation); //超时时间 inttimeout=getUrl().getMethodParameter(methodName,Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT); //单向调用 if(isOneway){ booleanisSent=getUrl().getMethodParameter(methodName,Constants.SENT_KEY,false); currentClient.send(inv,isSent); RpcContext.getContext().setFuture(null); returnnewRpcResult(); } //异步请求 elseif(isAsync){ ResponseFuturefuture=currentClient.request(inv,timeout); FutureAdapter<Object>futureAdapter=newFutureAdapter<>(future); RpcContext.getContext().setFuture(futureAdapter); Resultresult; if(isAsyncFuture){ result=newAsyncRpcResult(futureAdapter,futureAdapter.getResultFuture(),false); }else{ result=newSimpleAsyncRpcResult(futureAdapter,futureAdapter.getResultFuture(),false); } returnresult; } //同步请求 else{ RpcContext.getContext().setFuture(null); Resultresult=(Result)currentClient.request(inv,timeout).get(); returnresult; } }catch(TimeoutExceptione){ thrownewRpcException(RpcException.TIMEOUT_EXCEPTION,"Invokeremotemethodtimeout.method:"+invocation.getMethodName()+",provider:"+getUrl()+",cause:"+e.getMessage(),e); }catch(RemotingExceptione){ thrownewRpcException(RpcException.NETWORK_EXCEPTION,"Failedtoinvokeremotemethod:"+invocation.getMethodName()+",provider:"+getUrl()+",cause:"+e.getMessage(),e); } } } 我们看到与2.6.9版本相同的是FutureAdapter同样会被设置到上下文，但是FutureAdapter本身已经发生了变化： publicclassFutureAdapter<V>extendsCompletableFuture<V>{ privatefinalResponseFuturefuture; privateCompletableFuture<Result>resultFuture; publicFutureAdapter(ResponseFuturefuture){ this.future=future; this.resultFuture=newCompletableFuture<>(); //设置回调函数至DefaultFuture future.setCallback(newResponseCallback(){ //设置响应结果至CompletableFuture @Override publicvoiddone(Objectresponse){ Resultresult=(Result)response; FutureAdapter.this.resultFuture.complete(result); Vvalue=null; try{ value=(V)result.recreate(); }catch(Throwablet){ FutureAdapter.this.completeExceptionally(t); } FutureAdapter.this.complete(value); } //设置异常结果至FutureAdapter @Override publicvoidcaught(Throwableexception){ FutureAdapter.this.completeExceptionally(exception); } }); } publicResponseFuturegetFuture(){ returnfuture; } publicCompletableFuture<Result>getResultFuture(){ returnresultFuture; } } 我们在服务消费时通过getResultFuture方法获取CompletableFuture，这个对象值在回调时被设置，回调时机同样在DefaultFuture.doReceived方法里面： publicclassDefaultFutureimplementsResponseFuture{ privatevolatileResponseCallbackcallback; privatevoiddoReceived(Responseres){ lock.lock(); try{ response=res; if(done!=null){ done.signal(); } }finally{ lock.unlock(); } if(callback!=null){ //执行回调函数代码同version_2.6.9 invokeCallback(callback); } } } 4 文章总结 本文第一介绍了DUBBO消费异步化是什么，以及异步化为什么会带来性能提升。第二介绍了保护性暂停模式，这是实现异步化的基础。最后我们阅读了两个不同版本异步化源码，了解了DUBBO异步化演进过程，希望本文对大家有所帮助。 欢迎大家关注公众号「JAVA前线」查看更多精彩分享文章，主要包括源码分析、实际应用、架构思维、职场分享、产品思考等等，同时欢迎大家加我个人微信「java_front」一起交流学习
