1. 为什么技术路线图在国自然评审中至关重要第一次申报国自然时我把80%的精力都花在了研究背景和创新点上结果评审意见里赫然写着技术路线描述模糊可行性存疑。# 1. 概述本文分享Protocol 层。对应项目为dubbo-rpc和dubbo-rpc-api和dubbo-rpc-default。dubbo-rpc-api抽象API 层定义服务提供者Provider 和消费者Consumer 的行为。实现API 层实现代理、远程调用、监听过滤器等等功能。dubbo-rpc-default实现dubbo-rpc-api提供默认基于Dubbo协议的实现。本文涉及的类图如下2. RPC 核心接口2.1 Invokercom.alibaba.dubbo.rpc.Invoker调用接口。代码如下public interface InvokerT extends Node { /** * get service interface. * * 获得服务接口 * * return service interface. */ ClassT getInterface(); /** * invoke. * * 调用 * * param invocation * return result * throws RpcException */ Result invoke(Invocation invocation) throws RpcException; }一个 Invoker 是一个可执行体可根据名字协议方法信息参数发起调用。在服务提供者Invoker 用于调用服务提供类。在服务消费者Invoker 用于执行远程调用。一个 Invoker 是由ProxyFactory创建而来。2.1.1 InvokerListenercom.alibaba.dubbo.rpc.InvokerListenerInvoker 监听器。代码如下SPI public interface InvokerListener { /** * The invoker referred * * 当服务引用完成 * * param invoker * throws RpcException * see com.alibaba.dubbo.rpc.Protocol#refer(Class, com.alibaba.dubbo.common.URL) */ void referred(Invoker? invoker) throws RpcException; /** * The invoker destroyed. * * 当服务销毁引用 * * param invoker * throws RpcException * see com.alibaba.dubbo.rpc.Invoker#destroy() */ void destroyed(Invoker? invoker) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。每个方法上笔者添加了注释说明对应的事件。2.2 Protocolcom.alibaba.dubbo.rpc.Protocol协议接口。代码如下SPI(dubbo) public interface Protocol { /** * Get default port when user doesnt config the port. * * 获得默认的端口 * * return default port */ int getDefaultPort(); /** * Export service for remote invocation: br * 1. Protocol should record request source address after receive a request: * RpcContext.getContext().setRemoteAddress();br * 2. export() must be idempotent, that is, theres no difference between invoking once and invoking twice when * export the same URLbr * 3. Invoker instance is passed in by the framework, protocol needs not to care br * * 暴露远程服务 * 1. 协议在接收请求时应记录请求来源方地址信息RpcContext.getContext().setRemoteAddress();br * 2. export 必须是幂等的也就是暴露同一个 URL 的 Invoker 两次和暴露一次没有区别。br * 3. Invoker 由框架传入并协议解析协议不需要关心br * * param T Service type 服务的类型 * param invoker Service invoker 服务的执行体 * return exporter reference for exported service 暴露服务的引用用于取消暴露 * throws RpcException thrown when error occurs during export the service, e.g. port is occupied */ Adaptive T ExporterT export(InvokerT invoker) throws RpcException; /** * Refer a remote service: br * 1. When user calls invoke() method of Invoker object whichs returned from refer() call, the protocol * needs to correspondingly execute invoke() method of Invoker object br * 2. Its protocols responsibility to implement Invoker whichs returned from refer(). Generally speaking, * protocol sends remote request in the Invoker implementation. br * 3. When theres checkfalse set in URL, the implementation must not throw exception but try to recover when * connection fails. * * 引用远程服务 * 1. 当用户调用 refer 所返回 Invoker 对象的 invoke 方法时协议需相应执行同 URL 远端 export 传入的 Invoker 对象的 invoke 方法。br * 2. refer 返回的 Invoker 由协议实现协议通常需要在此 Invoker 中发送远程请求。br * 3. 当 URL 中有 checkfalse 参数时连接失败不能抛出异常并内部自动恢复。br * * param T Service type 服务的类型 * param type Service class 服务的类型 * param url URL address for the remote service 远程服务的URL地址 * return invoker services local proxy 服务的本地代理 * throws RpcException when theres any error while connecting to the service provider */ Adaptive T InvokerT refer(ClassT type, URL url) throws RpcException; /** * Destroy protocol: br * 1. Cancel all services this protocol exports and refers br * 2. Release all occupied resources, for example: connection, port, etc. br * 3. Protocol can continue to export and refer new service even after its destroyed. * * 释放协议 * 1. 取消该协议所有已经暴露和引用的服务。br * 2. 释放协议所占用的所有资源比如连接和端口。br * 3. 协议在释放后依然能暴露和引用新的服务。br */ void destroy(); }SPI(dubbo)注解Dubbo SPI拓展点默认为dubbo协议。Adaptive注解基于 Dubbo SPI Adaptive 机制加载对应的 Protocol 实现使用URL.protocol属性。#export(invoker)方法暴露远程服务。注意方法上的注释说明。#refer(type, url)方法引用远程服务。注意方法上的注释说明。#destroy()方法释放协议。2.2.1 ProtocolListenerWrappercom.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper实现 Protocol 接口协议监听器包装器实现类。代码如下public class ProtocolListenerWrapper implements Protocol { private final Protocol protocol; public ProtocolListenerWrapper(Protocol protocol) { if (protocol null) { throw new IllegalArgumentException(protocol null); } this.protocol protocol; } public int getDefaultPort() { return protocol.getDefaultPort(); } public T ExporterT export(InvokerT invoker) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 暴露服务创建 Exporter 对象 return new ListenerExporterWrapperT(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY))); } public T InvokerT refer(ClassT type, URL url) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } // 引用服务创建 Invoker 对象 return new ListenerInvokerWrapperT(protocol.refer(type, url), Collections.unmodifiableList( ExtensionLoader.getExtensionLoader(InvokerListener.class) .getActivateExtension(url, Constants.INVOKER_LISTENER_KEY))); } public void destroy() { protocol.destroy(); } }构造方法传入真正的 Protocol 对象例如DubboProtocol等等。#export(invoker)实现方法第 22 行当服务引用时注册中心的invoker.getUrl().getProtocol()返回的是registry。判断若是注册中心协议直接调用protocol#export(invoker)方法进行暴露服务。第 26 行创建 Exporter 对象ListenerExporterWrapper对象。通过ExporterListener监听服务的暴露事件。#refer(type, url)实现方法第 32 行当服务引用时注册中心的url.protocol返回的是registry。判断若是注册中心协议直接调用protocol#refer(type, url)方法进行引用服务。第 36 行创建 Invoker 对象ListenerInvokerWrapper对象。通过InvokerListener监听服务的引用事件。#destroy()实现方法调用protocol#destroy()方法进行销毁协议。2.2.2 ProtocolFilterWrappercom.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper实现 Protocol 接口协议过滤器包装器实现类。代码如下public class ProtocolFilterWrapper implements Protocol { private final Protocol protocol; public ProtocolFilterWrapper(Protocol protocol) { if (protocol null) { throw new IllegalArgumentException(protocol null); } this.protocol protocol; } public int getDefaultPort() { return protocol.getDefaultPort(); } public T ExporterT export(InvokerT invoker) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 暴露服务 return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER)); } public T InvokerT refer(ClassT type, URL url) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } // 引用服务 return buildInvokerChain(protocol.refer(type, url), Constants.REFERENCE_FILTER_KEY, Constants.CONSUMER); } public void destroy() { protocol.destroy(); } private static T InvokerT buildInvokerChain(final InvokerT invoker, String key, String group) { InvokerT last invoker; // 获得过滤器数组 ListFilter filters ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); // 倒序循环 Filter 创建带 Filter 链的 Invoker 对象 if (!filters.isEmpty()) { for (int i filters.size() - 1; i 0; i--) { final Filter filter filters.get(i); final InvokerT next last; last new InvokerT() { public ClassT getInterface() { return invoker.getInterface(); } public URL getUrl() { return invoker.getUrl(); } public boolean isAvailable() { return invoker.isAvailable(); } public Result invoke(Invocation invocation) throws RpcException { return filter.invoke(next, invocation); } public void destroy() { invoker.destroy(); } Override public String toString() { return invoker.toString(); } }; } } return last; } }#buildInvokerChain(invoker, key, group)静态方法创建带 Filter 链的 Invoker 对象。基于 Dubbo SPI Active 机制获得对应的过滤器数组。倒序循环 Filter 数组创建带 Filter 链的 Invoker 对象。注意虽然循环是倒序的但是最终得到的还是正序的 Filter 链。例如假设过滤器数组为[FilterA, FilterB]则得到的 Invoker 链为FilterA( FilterB( Invoker ) )。为什么是倒序呢我们在回过头看看ProtocolFilterWrapper#refer(type, url)方法创建的是服务消费者的 Invoker 对象。而 Filter 处理调用的顺序是FilterA FilterB Invoker即正序。也就是说虽然我们创建 Invoker 链是倒序但是最终执行是正序。为什么是这样的顺序例如通过 Filter 记录调用前和调用后的时间就可以计算调用耗时。所以Filter 是正序执行。再例如EchoFilter必须放在最后因此它的order -110000在META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Filter中配置。2.2.3 ExporterListenercom.alibaba.dubbo.rpc.ExporterListenerExporter 监听器。代码如下SPI public interface ExporterListener { /** * The exporter exported. * * 当服务暴露完成 * * param exporter * throws RpcException * see com.alibaba.dubbo.rpc.Protocol#export(Invoker) */ void exported(Exporter? exporter) throws RpcException; /** * The exporter unexported. * * 当服务取消暴露完成 * * param exporter * throws RpcException * see com.alibaba.dubbo.rpc.Exporter#unexport() */ void unexported(Exporter? exporter) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。每个方法上笔者添加了注释说明对应的事件。2.3 Exportercom.alibaba.dubbo.rpc.Exporter暴露接口。代码如下public interface ExporterT { /** * get invoker. * * 获得 Invoker 对象 * * return invoker */ InvokerT getInvoker(); /** * unexport. * * 取消暴露 * * throws RpcException */ void unexport(); }一个 Exporter 对应一个 Invoker 可通过#getInvoker()方法获得对应的 Invoker 对象。在服务提供者Protocol 暴露服务时会创建 Exporter 对象。2.4 Invocationcom.alibaba.dubbo.rpc.Invocation调用信息接口。代码如下public interface Invocation { /** * get method name. * * 获得方法名 * * return method name. * serial */ String getMethodName(); /** * get parameter types. * * 获得参数类型数组 * * return parameter types. * serial */ Class?[] getParameterTypes(); /** * get arguments. * * 获得参数值数组 * * return arguments. * serial */ Object[] getArguments(); /** * get attachments. * * 获得附加信息集合 * * return attachments. * serial */ MapString, String getAttachments(); /** * get attachment by key. * * 获得指定键的值 * * return attachment value. * serial */ String getAttachment(String key); /** * get attachment by key with default value. * * 获得指定键的值若不存在返回默认值 * * return attachment value. * serial */ String getAttachment(String key, String defaultValue); /** * get the invoker in current context. * * 获得 Invoker 对象 * * return invoker. * transient */ Invoker? getInvoker(); }在服务消费者使用 Invoker 对象发起 RPC 调用需要创建 Invocation 对象将调用信息方法参数等放入其中。在服务提供者会接收 Invocation 对象反射调用服务方法。2.5 Resultcom.alibaba.dubbo.rpc.Result结果接口。代码如下public interface Result { /** * Get invoke result. * * 获得返回值 * * return result. if no result return null. */ Object getValue(); /** * Get exception. * * 获得异常 * * return exception. if no exception return null. */ Throwable getException(); /** * Has exception. * * 是否有异常 * * return has exception. */ boolean hasException(); /** * Recreate. * p * code * if (hasException()) { * throw getException(); * } else { * return getValue(); * } * /code * * 重新抛出异常 * * return result. * throws Throwable exception */ Object recreate() throws Throwable; /** * get attachments. * * 获得附加信息集合 * * return attachments. */ MapString, String getAttachments(); /** * get attachment by key. * * 获得指定键的值 * * return attachment value. */ String getAttachment(String key); /** * get attachment by key with default value. * * 获得指定键的值若不存在返回默认值 * * return attachment value. */ String getAttachment(String key, String defaultValue); }在服务提供者使用 Invocation 对象反射调用服务方法返回 Result 对象。在服务消费者会接收 Result 对象返回值或者异常。2.5.1 RpcResultcom.alibaba.dubbo.rpc.RpcResult实现 Result 接口RPC 结果实现类。代码如下public class RpcResult implements Result, Serializable { private static final long serialVersionUID -6925924956850004727L; /** * 返回值 */ private Object result; /** * 异常 */ private Throwable exception; /** * 附加信息集合 */ private MapString, String attachments new HashMapString, String(); public RpcResult() { } public RpcResult(Object result) { this.result result; } public RpcResult(Throwable exception) { this.exception exception; } Override public Object recreate() throws Throwable { if (exception ! null) { // 抛出异常 throw exception; } return result; // 返回结果 } Override public Object getValue() { return result; } public void setValue(Object value) { this.result value; } Override public Throwable getException() { return exception; } public void setException(Throwable e) { this.exception e; } Override public boolean hasException() { return exception ! null; } Override public MapString, String getAttachments() { return attachments; } Override public String getAttachment(String key) { return attachments.get(key); } Override public String getAttachment(String key, String defaultValue) { String result attachments.get(key); if (result null || result.length() 0) { result defaultValue; } return result; } public void setAttachments(MapString, String map) { if (map ! null map.size() 0) { attachments.putAll(map); } } public void setAttachment(String key, String value) { attachments.put(key, value); } Override public String toString() { return RpcResult [result result , exception exception ]; } }代码比较简单胖友自己看注释。2.6 Filtercom.alibaba.dubbo.rpc.Filter过滤器接口。代码如下SPI public interface Filter { /** * do invoke filter. * p * code * // before filter * Result result invoker.invoke(invocation); * // after filter * return result; * /code * * 执行过滤逻辑 * * param invoker service * param invocation invocation. * return invoke result. * throws RpcException * see com.alibaba.dubbo.rpc.Invoker#invoke(Invocation) */ Result invoke(Invoker? invoker, Invocation invocation) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。在 Protocol 的实现类中会使用Filter构建调用链。例如ProtocolFilterWrapper#buildInvokerChain(invoker, key, group)方法。2.6.1 ActiveLimitFiltercom.alibaba.dubbo.rpc.filter.ActiveLimitFilter实现 Filter 接口方法的最大可并行调用的限制过滤器。代码如下Activate(group Constants.CONSUMER, value Constants.ACTIVES_KEY) public class ActiveLimitFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 获得 URL 对象 URL url invoker.getUrl(); // 获得方法名 String methodName invocation.getMethodName(); // 获得最大可并行调用数 int max invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0); // 获得 RpcStatus 对象基于服务 URL 方法维度 RpcStatus count RpcStatus.getStatus(url, invocation.getMethodName()); if (max 0) { // 超过最大可并行调用数 // 获得超时值 long timeout invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0); long start System.currentTimeMillis(); long remain timeout; int active count.getActive(); if (active max) { // 超过最大可并行调用数 synchronized (count) { // 通过锁有且仅有一个在等待。 while ((active count.getActive()) max) { try { count.wait(remain); } catch (InterruptedException e) { } // 判断是否没有剩余时长了抛出 RpcException 异常 long elapsed System.currentTimeMillis() - start; remain timeout - elapsed; if (remain 0) { throw new RpcException(Waiting concurrent invoke timeout in client-side for service: invoker.getInterface().getName() , method: invocation.getMethodName() , elapsed: elapsed , timeout: timeout . concurrent invokes: active . max concurrent invoke limit: max); } } } } } try { // 调用开始的计数 long begin System.currentTimeMillis(); RpcStatus.beginCount(url, methodName); try { // 调用服务 Result result invoker.invoke(invocation); // 调用结束的计数成功 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true); return result; } catch (RuntimeException t) { // 调用结束的计数失败 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false); throw t; } } finally { // 唤醒等待的相同服务的相同方法的请求 if (max 0) { synchronized (count) { count.notifyAll(); } } } } }Activate(group Constants.CONSUMER, value Constants.ACTIVES_KEY)注解基于服务消费者的方法的最大可并行调用的限制过滤器。RpcStatus 记录当前服务或者服务的方法的各种统计信息包括调用次数、失败次数、成功次数等等。通过它可以实现在#invoke(invoker, invocation)方法中方法的调用数的判断。 代码比较简单胖友自己看注释。2.6.2 EchoFiltercom.alibaba.dubbo.rpc.filter.EchoFilter实现 Filter 接口回声测试过滤器。代码如下Activate(group Constants.PROVIDER, order -110000) public class EchoFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation inv) throws RpcException { // 方法名为 $echo 参数只有一个 if (inv.getMethodName().equals(Constants.$ECHO) inv.getArguments() ! null inv.getArguments().length 1) { return new RpcResult(inv.getArguments()[0]); } return invoker.invoke(inv); } }Activate(group Constants.PROVIDER, order -110000)注解基于服务提供者的回声测试过滤器。回声测试方法的方法名为$echo参数只有一个直接返回该参数。因为不需要调用服务提供者所以直接返回结果。2.6.3 ExceptionFiltercom.alibaba.dubbo.rpc.filter.ExceptionFilter实现 Filter 接口异常过滤器。代码如下Activate(group Constants.PROVIDER) public class ExceptionFilter implements Filter { private final Logger logger; public ExceptionFilter() { this(LoggerFactory.getLogger(ExceptionFilter.class)); } public ExceptionFilter(Logger logger) { this.logger logger; } Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { try { // 服务调用 Result result invoker.invoke(invocation); // 有异常并且非泛化调用 if (result.hasException() GenericService.class ! invoker.getInterface()) { try { Throwable exception result.getException(); // directly throw if its checked exception // 如果是 checked 异常直接抛出 if (!(exception instanceof RuntimeException) (exception instanceof Exception)) { return result; } // directly throw if the exception appears in the signature // 在方法签名上有声明直接抛出 try { Method method invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes()); Class?[] exceptionClassses method.getExceptionTypes(); for (Class? exceptionClass : exceptionClassses) { if (exception.getClass().equals(exceptionClass)) { return result; } } } catch (NoSuchMethodException e) { return result; } // 未在方法签名上定义的异常在服务器端打印 ERROR 日志 // for the exception not found in methods signature, print ERROR message in servers log. logger.error(Got unchecked and undeclared exception which called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: exception.getClass().getName() : exception.getMessage(), exception); // 异常类和接口类在同一 jar 包里直接抛出 // directly throw if exception class and interface class are in the same jar file. String serviceFile ReflectUtils.getCodeBase(invoker.getInterface()); String exceptionFile ReflectUtils.getCodeBase(exception.getClass()); if (serviceFile null || exceptionFile null || serviceFile.equals(exceptionFile)) { return result; } // 是JDK自带的异常直接抛出 // directly throw if its JDK exception String className exception.getClass().getName(); if (className.startsWith(java.) || className.startsWith(javax.)) { return result; } // 是Dubbo本身的异常直接抛出 // directly throw if its dubbo exception if (exception instanceof RpcException) { return result; } // 否则包装成RuntimeException抛给客户端 // otherwise, wrap with RuntimeException and throw back to the client return new RpcResult(new RuntimeException(StringUtils.toString(exception))); } catch (Throwable e) { logger.warn(Fail to ExceptionFilter when called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: e.getClass().getName() : e.getMessage(), e); return result; } } return result; } catch (RuntimeException e) { logger.error(Got unchecked and undeclared exception which called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: e.getClass().getName() : e.getMessage(), e); throw e; } } }Activate(group Constants.PROVIDER)注解基于服务提供者的异常过滤器。目的是不将服务提供者的异常直接抛给服务消费者而是包装后返回。同时在服务提供者一侧打印错误日志。 代码比较简单胖友自己看注释。2.6.4 TimeoutFiltercom.alibaba.dubbo.rpc.filter.TimeoutFilter实现 Filter 接口超时过滤器。代码如下Activate(group Constants.PROVIDER) public class TimeoutFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 调用开始时间 long start System.currentTimeMillis(); // 服务调用 Result result invoker.invoke(invocation); // 计算调用时长 long elapsed System.currentTimeMillis() - start; // 超过时长打印告警日志 if (invoker.getUrl() ! null elapsed invoker.getUrl().getMethodParameter(invocation.getMethodName(), timeout, Integer.MAX_VALUE)) { if (logger.isWarnEnabled()) { logger.warn(invoke time out. method: invocation.getMethodName() arguments: Arrays.toString(invocation.getArguments()) , url is invoker.getUrl() , invoke elapsed elapsed ms.); } } return result; } }Activate(group Constants.PROVIDER)注解基于服务提供者的超时过滤器。目的是记录服务提供者的超时方法的调用打印告警日志。2.6.5 ExecuteLimitFiltercom.alibaba.dubbo.rpc.filter.ExecuteLimitFilter实现 Filter 接口方法的最大可并行执行数的限制过滤器。代码如下Activate(group Constants.PROVIDER, value Constants.EXECUTES_KEY) public class ExecuteLimitFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 获得 URL 对象 URL url invoker.getUrl(); // 方法名 String methodName invocation.getMethodName(); // 并发执行数 int max url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0); // 超过限制抛出 RpcException 异常 if (max 0) { RpcStatus count RpcStatus.getStatus(url, invocation.getMethodName()); if (count.getActive() max) { throw new RpcException(Failed to invoke method invocation.getMethodName() in provider url , cause: The service using threads greater than dubbo:service executes\ max \ / limited.); } } // 调用开始的计数 long begin System.currentTimeMillis(); boolean isSuccess true; RpcStatus.beginCount(url, methodName); try { // 服务调用 Result result invoker.invoke(invocation); return result; } catch (Throwable t) { isSuccess false; if (t instanceof RuntimeException) { throw (RuntimeException) t; } else { throw new RpcException(unexpected exception when ExecuteLimitFilter, t); } } finally { // 调用结束的计数成功失败 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess); } } }Activate(group Constants.PROVIDER, value Constants.EXECUTES_KEY)注解基于服务提供者的方法的最大可并行执行数的限制过滤器。和 ActiveLimitFilter 有一些相似点胖友可以自己对比下。2.6.6 GenericFiltercom.alibaba.dubbo.rpc.filter.GenericFilter实现 Filter 接口泛化调用过滤器。代码如下Activate(group Dubbo RPC核心接口与过滤器机制解析
1. 为什么技术路线图在国自然评审中至关重要第一次申报国自然时我把80%的精力都花在了研究背景和创新点上结果评审意见里赫然写着技术路线描述模糊可行性存疑。# 1. 概述本文分享Protocol 层。对应项目为dubbo-rpc和dubbo-rpc-api和dubbo-rpc-default。dubbo-rpc-api抽象API 层定义服务提供者Provider 和消费者Consumer 的行为。实现API 层实现代理、远程调用、监听过滤器等等功能。dubbo-rpc-default实现dubbo-rpc-api提供默认基于Dubbo协议的实现。本文涉及的类图如下2. RPC 核心接口2.1 Invokercom.alibaba.dubbo.rpc.Invoker调用接口。代码如下public interface InvokerT extends Node { /** * get service interface. * * 获得服务接口 * * return service interface. */ ClassT getInterface(); /** * invoke. * * 调用 * * param invocation * return result * throws RpcException */ Result invoke(Invocation invocation) throws RpcException; }一个 Invoker 是一个可执行体可根据名字协议方法信息参数发起调用。在服务提供者Invoker 用于调用服务提供类。在服务消费者Invoker 用于执行远程调用。一个 Invoker 是由ProxyFactory创建而来。2.1.1 InvokerListenercom.alibaba.dubbo.rpc.InvokerListenerInvoker 监听器。代码如下SPI public interface InvokerListener { /** * The invoker referred * * 当服务引用完成 * * param invoker * throws RpcException * see com.alibaba.dubbo.rpc.Protocol#refer(Class, com.alibaba.dubbo.common.URL) */ void referred(Invoker? invoker) throws RpcException; /** * The invoker destroyed. * * 当服务销毁引用 * * param invoker * throws RpcException * see com.alibaba.dubbo.rpc.Invoker#destroy() */ void destroyed(Invoker? invoker) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。每个方法上笔者添加了注释说明对应的事件。2.2 Protocolcom.alibaba.dubbo.rpc.Protocol协议接口。代码如下SPI(dubbo) public interface Protocol { /** * Get default port when user doesnt config the port. * * 获得默认的端口 * * return default port */ int getDefaultPort(); /** * Export service for remote invocation: br * 1. Protocol should record request source address after receive a request: * RpcContext.getContext().setRemoteAddress();br * 2. export() must be idempotent, that is, theres no difference between invoking once and invoking twice when * export the same URLbr * 3. Invoker instance is passed in by the framework, protocol needs not to care br * * 暴露远程服务 * 1. 协议在接收请求时应记录请求来源方地址信息RpcContext.getContext().setRemoteAddress();br * 2. export 必须是幂等的也就是暴露同一个 URL 的 Invoker 两次和暴露一次没有区别。br * 3. Invoker 由框架传入并协议解析协议不需要关心br * * param T Service type 服务的类型 * param invoker Service invoker 服务的执行体 * return exporter reference for exported service 暴露服务的引用用于取消暴露 * throws RpcException thrown when error occurs during export the service, e.g. port is occupied */ Adaptive T ExporterT export(InvokerT invoker) throws RpcException; /** * Refer a remote service: br * 1. When user calls invoke() method of Invoker object whichs returned from refer() call, the protocol * needs to correspondingly execute invoke() method of Invoker object br * 2. Its protocols responsibility to implement Invoker whichs returned from refer(). Generally speaking, * protocol sends remote request in the Invoker implementation. br * 3. When theres checkfalse set in URL, the implementation must not throw exception but try to recover when * connection fails. * * 引用远程服务 * 1. 当用户调用 refer 所返回 Invoker 对象的 invoke 方法时协议需相应执行同 URL 远端 export 传入的 Invoker 对象的 invoke 方法。br * 2. refer 返回的 Invoker 由协议实现协议通常需要在此 Invoker 中发送远程请求。br * 3. 当 URL 中有 checkfalse 参数时连接失败不能抛出异常并内部自动恢复。br * * param T Service type 服务的类型 * param type Service class 服务的类型 * param url URL address for the remote service 远程服务的URL地址 * return invoker services local proxy 服务的本地代理 * throws RpcException when theres any error while connecting to the service provider */ Adaptive T InvokerT refer(ClassT type, URL url) throws RpcException; /** * Destroy protocol: br * 1. Cancel all services this protocol exports and refers br * 2. Release all occupied resources, for example: connection, port, etc. br * 3. Protocol can continue to export and refer new service even after its destroyed. * * 释放协议 * 1. 取消该协议所有已经暴露和引用的服务。br * 2. 释放协议所占用的所有资源比如连接和端口。br * 3. 协议在释放后依然能暴露和引用新的服务。br */ void destroy(); }SPI(dubbo)注解Dubbo SPI拓展点默认为dubbo协议。Adaptive注解基于 Dubbo SPI Adaptive 机制加载对应的 Protocol 实现使用URL.protocol属性。#export(invoker)方法暴露远程服务。注意方法上的注释说明。#refer(type, url)方法引用远程服务。注意方法上的注释说明。#destroy()方法释放协议。2.2.1 ProtocolListenerWrappercom.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper实现 Protocol 接口协议监听器包装器实现类。代码如下public class ProtocolListenerWrapper implements Protocol { private final Protocol protocol; public ProtocolListenerWrapper(Protocol protocol) { if (protocol null) { throw new IllegalArgumentException(protocol null); } this.protocol protocol; } public int getDefaultPort() { return protocol.getDefaultPort(); } public T ExporterT export(InvokerT invoker) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 暴露服务创建 Exporter 对象 return new ListenerExporterWrapperT(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY))); } public T InvokerT refer(ClassT type, URL url) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } // 引用服务创建 Invoker 对象 return new ListenerInvokerWrapperT(protocol.refer(type, url), Collections.unmodifiableList( ExtensionLoader.getExtensionLoader(InvokerListener.class) .getActivateExtension(url, Constants.INVOKER_LISTENER_KEY))); } public void destroy() { protocol.destroy(); } }构造方法传入真正的 Protocol 对象例如DubboProtocol等等。#export(invoker)实现方法第 22 行当服务引用时注册中心的invoker.getUrl().getProtocol()返回的是registry。判断若是注册中心协议直接调用protocol#export(invoker)方法进行暴露服务。第 26 行创建 Exporter 对象ListenerExporterWrapper对象。通过ExporterListener监听服务的暴露事件。#refer(type, url)实现方法第 32 行当服务引用时注册中心的url.protocol返回的是registry。判断若是注册中心协议直接调用protocol#refer(type, url)方法进行引用服务。第 36 行创建 Invoker 对象ListenerInvokerWrapper对象。通过InvokerListener监听服务的引用事件。#destroy()实现方法调用protocol#destroy()方法进行销毁协议。2.2.2 ProtocolFilterWrappercom.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper实现 Protocol 接口协议过滤器包装器实现类。代码如下public class ProtocolFilterWrapper implements Protocol { private final Protocol protocol; public ProtocolFilterWrapper(Protocol protocol) { if (protocol null) { throw new IllegalArgumentException(protocol null); } this.protocol protocol; } public int getDefaultPort() { return protocol.getDefaultPort(); } public T ExporterT export(InvokerT invoker) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } // 暴露服务 return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER)); } public T InvokerT refer(ClassT type, URL url) throws RpcException { // 注册中心 if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } // 引用服务 return buildInvokerChain(protocol.refer(type, url), Constants.REFERENCE_FILTER_KEY, Constants.CONSUMER); } public void destroy() { protocol.destroy(); } private static T InvokerT buildInvokerChain(final InvokerT invoker, String key, String group) { InvokerT last invoker; // 获得过滤器数组 ListFilter filters ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); // 倒序循环 Filter 创建带 Filter 链的 Invoker 对象 if (!filters.isEmpty()) { for (int i filters.size() - 1; i 0; i--) { final Filter filter filters.get(i); final InvokerT next last; last new InvokerT() { public ClassT getInterface() { return invoker.getInterface(); } public URL getUrl() { return invoker.getUrl(); } public boolean isAvailable() { return invoker.isAvailable(); } public Result invoke(Invocation invocation) throws RpcException { return filter.invoke(next, invocation); } public void destroy() { invoker.destroy(); } Override public String toString() { return invoker.toString(); } }; } } return last; } }#buildInvokerChain(invoker, key, group)静态方法创建带 Filter 链的 Invoker 对象。基于 Dubbo SPI Active 机制获得对应的过滤器数组。倒序循环 Filter 数组创建带 Filter 链的 Invoker 对象。注意虽然循环是倒序的但是最终得到的还是正序的 Filter 链。例如假设过滤器数组为[FilterA, FilterB]则得到的 Invoker 链为FilterA( FilterB( Invoker ) )。为什么是倒序呢我们在回过头看看ProtocolFilterWrapper#refer(type, url)方法创建的是服务消费者的 Invoker 对象。而 Filter 处理调用的顺序是FilterA FilterB Invoker即正序。也就是说虽然我们创建 Invoker 链是倒序但是最终执行是正序。为什么是这样的顺序例如通过 Filter 记录调用前和调用后的时间就可以计算调用耗时。所以Filter 是正序执行。再例如EchoFilter必须放在最后因此它的order -110000在META-INF/dubbo/internal/com.alibaba.dubbo.rpc.Filter中配置。2.2.3 ExporterListenercom.alibaba.dubbo.rpc.ExporterListenerExporter 监听器。代码如下SPI public interface ExporterListener { /** * The exporter exported. * * 当服务暴露完成 * * param exporter * throws RpcException * see com.alibaba.dubbo.rpc.Protocol#export(Invoker) */ void exported(Exporter? exporter) throws RpcException; /** * The exporter unexported. * * 当服务取消暴露完成 * * param exporter * throws RpcException * see com.alibaba.dubbo.rpc.Exporter#unexport() */ void unexported(Exporter? exporter) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。每个方法上笔者添加了注释说明对应的事件。2.3 Exportercom.alibaba.dubbo.rpc.Exporter暴露接口。代码如下public interface ExporterT { /** * get invoker. * * 获得 Invoker 对象 * * return invoker */ InvokerT getInvoker(); /** * unexport. * * 取消暴露 * * throws RpcException */ void unexport(); }一个 Exporter 对应一个 Invoker 可通过#getInvoker()方法获得对应的 Invoker 对象。在服务提供者Protocol 暴露服务时会创建 Exporter 对象。2.4 Invocationcom.alibaba.dubbo.rpc.Invocation调用信息接口。代码如下public interface Invocation { /** * get method name. * * 获得方法名 * * return method name. * serial */ String getMethodName(); /** * get parameter types. * * 获得参数类型数组 * * return parameter types. * serial */ Class?[] getParameterTypes(); /** * get arguments. * * 获得参数值数组 * * return arguments. * serial */ Object[] getArguments(); /** * get attachments. * * 获得附加信息集合 * * return attachments. * serial */ MapString, String getAttachments(); /** * get attachment by key. * * 获得指定键的值 * * return attachment value. * serial */ String getAttachment(String key); /** * get attachment by key with default value. * * 获得指定键的值若不存在返回默认值 * * return attachment value. * serial */ String getAttachment(String key, String defaultValue); /** * get the invoker in current context. * * 获得 Invoker 对象 * * return invoker. * transient */ Invoker? getInvoker(); }在服务消费者使用 Invoker 对象发起 RPC 调用需要创建 Invocation 对象将调用信息方法参数等放入其中。在服务提供者会接收 Invocation 对象反射调用服务方法。2.5 Resultcom.alibaba.dubbo.rpc.Result结果接口。代码如下public interface Result { /** * Get invoke result. * * 获得返回值 * * return result. if no result return null. */ Object getValue(); /** * Get exception. * * 获得异常 * * return exception. if no exception return null. */ Throwable getException(); /** * Has exception. * * 是否有异常 * * return has exception. */ boolean hasException(); /** * Recreate. * p * code * if (hasException()) { * throw getException(); * } else { * return getValue(); * } * /code * * 重新抛出异常 * * return result. * throws Throwable exception */ Object recreate() throws Throwable; /** * get attachments. * * 获得附加信息集合 * * return attachments. */ MapString, String getAttachments(); /** * get attachment by key. * * 获得指定键的值 * * return attachment value. */ String getAttachment(String key); /** * get attachment by key with default value. * * 获得指定键的值若不存在返回默认值 * * return attachment value. */ String getAttachment(String key, String defaultValue); }在服务提供者使用 Invocation 对象反射调用服务方法返回 Result 对象。在服务消费者会接收 Result 对象返回值或者异常。2.5.1 RpcResultcom.alibaba.dubbo.rpc.RpcResult实现 Result 接口RPC 结果实现类。代码如下public class RpcResult implements Result, Serializable { private static final long serialVersionUID -6925924956850004727L; /** * 返回值 */ private Object result; /** * 异常 */ private Throwable exception; /** * 附加信息集合 */ private MapString, String attachments new HashMapString, String(); public RpcResult() { } public RpcResult(Object result) { this.result result; } public RpcResult(Throwable exception) { this.exception exception; } Override public Object recreate() throws Throwable { if (exception ! null) { // 抛出异常 throw exception; } return result; // 返回结果 } Override public Object getValue() { return result; } public void setValue(Object value) { this.result value; } Override public Throwable getException() { return exception; } public void setException(Throwable e) { this.exception e; } Override public boolean hasException() { return exception ! null; } Override public MapString, String getAttachments() { return attachments; } Override public String getAttachment(String key) { return attachments.get(key); } Override public String getAttachment(String key, String defaultValue) { String result attachments.get(key); if (result null || result.length() 0) { result defaultValue; } return result; } public void setAttachments(MapString, String map) { if (map ! null map.size() 0) { attachments.putAll(map); } } public void setAttachment(String key, String value) { attachments.put(key, value); } Override public String toString() { return RpcResult [result result , exception exception ]; } }代码比较简单胖友自己看注释。2.6 Filtercom.alibaba.dubbo.rpc.Filter过滤器接口。代码如下SPI public interface Filter { /** * do invoke filter. * p * code * // before filter * Result result invoker.invoke(invocation); * // after filter * return result; * /code * * 执行过滤逻辑 * * param invoker service * param invocation invocation. * return invoke result. * throws RpcException * see com.alibaba.dubbo.rpc.Invoker#invoke(Invocation) */ Result invoke(Invoker? invoker, Invocation invocation) throws RpcException; }SPI注解Dubbo SPI拓展点无默认值。在 Protocol 的实现类中会使用Filter构建调用链。例如ProtocolFilterWrapper#buildInvokerChain(invoker, key, group)方法。2.6.1 ActiveLimitFiltercom.alibaba.dubbo.rpc.filter.ActiveLimitFilter实现 Filter 接口方法的最大可并行调用的限制过滤器。代码如下Activate(group Constants.CONSUMER, value Constants.ACTIVES_KEY) public class ActiveLimitFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 获得 URL 对象 URL url invoker.getUrl(); // 获得方法名 String methodName invocation.getMethodName(); // 获得最大可并行调用数 int max invoker.getUrl().getMethodParameter(methodName, Constants.ACTIVES_KEY, 0); // 获得 RpcStatus 对象基于服务 URL 方法维度 RpcStatus count RpcStatus.getStatus(url, invocation.getMethodName()); if (max 0) { // 超过最大可并行调用数 // 获得超时值 long timeout invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.TIMEOUT_KEY, 0); long start System.currentTimeMillis(); long remain timeout; int active count.getActive(); if (active max) { // 超过最大可并行调用数 synchronized (count) { // 通过锁有且仅有一个在等待。 while ((active count.getActive()) max) { try { count.wait(remain); } catch (InterruptedException e) { } // 判断是否没有剩余时长了抛出 RpcException 异常 long elapsed System.currentTimeMillis() - start; remain timeout - elapsed; if (remain 0) { throw new RpcException(Waiting concurrent invoke timeout in client-side for service: invoker.getInterface().getName() , method: invocation.getMethodName() , elapsed: elapsed , timeout: timeout . concurrent invokes: active . max concurrent invoke limit: max); } } } } } try { // 调用开始的计数 long begin System.currentTimeMillis(); RpcStatus.beginCount(url, methodName); try { // 调用服务 Result result invoker.invoke(invocation); // 调用结束的计数成功 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, true); return result; } catch (RuntimeException t) { // 调用结束的计数失败 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, false); throw t; } } finally { // 唤醒等待的相同服务的相同方法的请求 if (max 0) { synchronized (count) { count.notifyAll(); } } } } }Activate(group Constants.CONSUMER, value Constants.ACTIVES_KEY)注解基于服务消费者的方法的最大可并行调用的限制过滤器。RpcStatus 记录当前服务或者服务的方法的各种统计信息包括调用次数、失败次数、成功次数等等。通过它可以实现在#invoke(invoker, invocation)方法中方法的调用数的判断。 代码比较简单胖友自己看注释。2.6.2 EchoFiltercom.alibaba.dubbo.rpc.filter.EchoFilter实现 Filter 接口回声测试过滤器。代码如下Activate(group Constants.PROVIDER, order -110000) public class EchoFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation inv) throws RpcException { // 方法名为 $echo 参数只有一个 if (inv.getMethodName().equals(Constants.$ECHO) inv.getArguments() ! null inv.getArguments().length 1) { return new RpcResult(inv.getArguments()[0]); } return invoker.invoke(inv); } }Activate(group Constants.PROVIDER, order -110000)注解基于服务提供者的回声测试过滤器。回声测试方法的方法名为$echo参数只有一个直接返回该参数。因为不需要调用服务提供者所以直接返回结果。2.6.3 ExceptionFiltercom.alibaba.dubbo.rpc.filter.ExceptionFilter实现 Filter 接口异常过滤器。代码如下Activate(group Constants.PROVIDER) public class ExceptionFilter implements Filter { private final Logger logger; public ExceptionFilter() { this(LoggerFactory.getLogger(ExceptionFilter.class)); } public ExceptionFilter(Logger logger) { this.logger logger; } Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { try { // 服务调用 Result result invoker.invoke(invocation); // 有异常并且非泛化调用 if (result.hasException() GenericService.class ! invoker.getInterface()) { try { Throwable exception result.getException(); // directly throw if its checked exception // 如果是 checked 异常直接抛出 if (!(exception instanceof RuntimeException) (exception instanceof Exception)) { return result; } // directly throw if the exception appears in the signature // 在方法签名上有声明直接抛出 try { Method method invoker.getInterface().getMethod(invocation.getMethodName(), invocation.getParameterTypes()); Class?[] exceptionClassses method.getExceptionTypes(); for (Class? exceptionClass : exceptionClassses) { if (exception.getClass().equals(exceptionClass)) { return result; } } } catch (NoSuchMethodException e) { return result; } // 未在方法签名上定义的异常在服务器端打印 ERROR 日志 // for the exception not found in methods signature, print ERROR message in servers log. logger.error(Got unchecked and undeclared exception which called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: exception.getClass().getName() : exception.getMessage(), exception); // 异常类和接口类在同一 jar 包里直接抛出 // directly throw if exception class and interface class are in the same jar file. String serviceFile ReflectUtils.getCodeBase(invoker.getInterface()); String exceptionFile ReflectUtils.getCodeBase(exception.getClass()); if (serviceFile null || exceptionFile null || serviceFile.equals(exceptionFile)) { return result; } // 是JDK自带的异常直接抛出 // directly throw if its JDK exception String className exception.getClass().getName(); if (className.startsWith(java.) || className.startsWith(javax.)) { return result; } // 是Dubbo本身的异常直接抛出 // directly throw if its dubbo exception if (exception instanceof RpcException) { return result; } // 否则包装成RuntimeException抛给客户端 // otherwise, wrap with RuntimeException and throw back to the client return new RpcResult(new RuntimeException(StringUtils.toString(exception))); } catch (Throwable e) { logger.warn(Fail to ExceptionFilter when called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: e.getClass().getName() : e.getMessage(), e); return result; } } return result; } catch (RuntimeException e) { logger.error(Got unchecked and undeclared exception which called by RpcContext.getContext().getRemoteHost() . service: invoker.getInterface().getName() , method: invocation.getMethodName() , exception: e.getClass().getName() : e.getMessage(), e); throw e; } } }Activate(group Constants.PROVIDER)注解基于服务提供者的异常过滤器。目的是不将服务提供者的异常直接抛给服务消费者而是包装后返回。同时在服务提供者一侧打印错误日志。 代码比较简单胖友自己看注释。2.6.4 TimeoutFiltercom.alibaba.dubbo.rpc.filter.TimeoutFilter实现 Filter 接口超时过滤器。代码如下Activate(group Constants.PROVIDER) public class TimeoutFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 调用开始时间 long start System.currentTimeMillis(); // 服务调用 Result result invoker.invoke(invocation); // 计算调用时长 long elapsed System.currentTimeMillis() - start; // 超过时长打印告警日志 if (invoker.getUrl() ! null elapsed invoker.getUrl().getMethodParameter(invocation.getMethodName(), timeout, Integer.MAX_VALUE)) { if (logger.isWarnEnabled()) { logger.warn(invoke time out. method: invocation.getMethodName() arguments: Arrays.toString(invocation.getArguments()) , url is invoker.getUrl() , invoke elapsed elapsed ms.); } } return result; } }Activate(group Constants.PROVIDER)注解基于服务提供者的超时过滤器。目的是记录服务提供者的超时方法的调用打印告警日志。2.6.5 ExecuteLimitFiltercom.alibaba.dubbo.rpc.filter.ExecuteLimitFilter实现 Filter 接口方法的最大可并行执行数的限制过滤器。代码如下Activate(group Constants.PROVIDER, value Constants.EXECUTES_KEY) public class ExecuteLimitFilter implements Filter { Override public Result invoke(Invoker? invoker, Invocation invocation) throws RpcException { // 获得 URL 对象 URL url invoker.getUrl(); // 方法名 String methodName invocation.getMethodName(); // 并发执行数 int max url.getMethodParameter(methodName, Constants.EXECUTES_KEY, 0); // 超过限制抛出 RpcException 异常 if (max 0) { RpcStatus count RpcStatus.getStatus(url, invocation.getMethodName()); if (count.getActive() max) { throw new RpcException(Failed to invoke method invocation.getMethodName() in provider url , cause: The service using threads greater than dubbo:service executes\ max \ / limited.); } } // 调用开始的计数 long begin System.currentTimeMillis(); boolean isSuccess true; RpcStatus.beginCount(url, methodName); try { // 服务调用 Result result invoker.invoke(invocation); return result; } catch (Throwable t) { isSuccess false; if (t instanceof RuntimeException) { throw (RuntimeException) t; } else { throw new RpcException(unexpected exception when ExecuteLimitFilter, t); } } finally { // 调用结束的计数成功失败 RpcStatus.endCount(url, methodName, System.currentTimeMillis() - begin, isSuccess); } } }Activate(group Constants.PROVIDER, value Constants.EXECUTES_KEY)注解基于服务提供者的方法的最大可并行执行数的限制过滤器。和 ActiveLimitFilter 有一些相似点胖友可以自己对比下。2.6.6 GenericFiltercom.alibaba.dubbo.rpc.filter.GenericFilter实现 Filter 接口泛化调用过滤器。代码如下Activate(group