Dubbo 是阿里巴巴开源的一个高性能优秀的服务框架,使得应用可通过高性能的 RPC 实现服务的输入与输出功能。作为一个优秀的框架,至少应该包含以下几个特点:
完善的文档
活跃的社区
良好的扩展性
今天主要讨论的主题就是 dubbo 中良好的扩展性。 dubbo 的扩展点加载是从 JDK 标准的 SPI (Service Provider Interface) 扩展点发现加强而来。
JDK 标准的 SPI 会一次性实例化扩展点所有实现,如果有扩展实现初始化很耗时,但如果没用上也加载,会很浪费资源。
如果扩展点加载失败,连扩展点的名称都拿不到了。比如:JDK 标准的 ScriptEngine,通过 getName() 获取脚本类型的名称,但如果 RubyScriptEngine 因为所依赖的 jruby.jar 不存在,导致 RubyScriptEngine 类加载失败,这个失败原因被吃掉了,和 ruby 对应不起来,当用户执行 ruby 脚本时,会报不支持 ruby,而不是真正失败的原因。
增加了对扩展点 IoC 和 AOP 的支持,一个扩展点可以直接 setter 注入其它扩展点。
我们知道 dubbo 最核心的三个概念分别是:Provider(服务提供者)、Consumer(服务消费者)与 Registry(注册中心),通过注册中心解耦了服务方与消费方的调用关系。在服务发布与服务引用的时候 dubbo 分别提供了 ExporterListener 与 InvokerListener 对于服务进行不同的扩展。
dubbo 在进行服务调用的过程中最核心的概念就是 Invoke,就相当于 Spring 里面的 bean 一样。对于 Invoke 这个核心模型 dubbo 也有 Filter 对其进行扩展。在这里大家有没有联想到 Servlet 里面的 Listener 与 Filter,可以看到优秀的框架里面有很多思想都是相通的。
在 dubbo 中主要包含以下三个核心领域模型:
dubbo 通过 Protocol 服务暴露 Invoke 这个服务调用的执行体,然后通过 Protocol 引用 Invoke 调用 Invocation 提供调用过程中的变量就可以完成一次服务的发布与调用过程。
dubbo 的服务发布与引用都提供了Listener 分别是 ExporterListener(服务暴露监听) 与 InvokerListener(服务调用监听)。
ExporterListener 是 dubbo 通过 Protocol 进行服务暴露的时候调用的扩展点,也就是 Protocol 在进行 export() 方法的时候对服务暴露的一个扩展点。最终是在 ProtocolListenerWrapper#export 方法中通过 dubbo 的 SPI 机制加载进去,然后包装成一个 ListenerExporterWrapper 对象。
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } return new ListenerExporterWrapper<T>(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY))); }
ExporterListener 接口的定义如下:
public interface ExporterListener { void exported(Exporter<?> exporter) throws RpcException; void unexported(Exporter<?> exporter); }
InvokerListener是 dubbo 通过 Protocol 进行服务引用的时候调用的扩展点,也就是 Protocol 在进行 refer() 方法的时候对服务暴露的一个扩展点。最终是在 ProtocolListenerWrapper#export 方法中通过 dubbo 的 SPI 机制加载进去,然后包装成一个 ListenerInvokerWrapper 对象。
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) { return protocol.refer(type, url); } return new ListenerInvokerWrapper<T>(protocol.refer(type, url), Collections.unmodifiableList( ExtensionLoader.getExtensionLoader(InvokerListener.class) .getActivateExtension(url, Constants.INVOKER_LISTENER_KEY))); }
InvokerListener 接口的定义如下:
public interface InvokerListener { void referred(Invoker<?> invoker) throws RpcException; void destroyed(Invoker<?> invoker); }
Filter 是 dubbo 对于服务调用 (Invoker) 的进行拦截。dubbo 中 Invoke 是服务暴露方与服务引用方共用的一个核心概念,所以对于 Filter 对于这两都都会进行拦截。分别通过 ProtocolFilterWrapper#export 与 ProtocolFilterWrapper#refer 构建 Filter 对服务暴露方与服务引用方的 Invoke 进行拦截。
1) 服务暴露方
public <T> Exporter<T> export(Invoker<T> 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)); }
2) 服务引用方
public <T> Invoker<T> refer(Class<T> 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); }
服务暴露方与引用方它们都是构建成一个或者多个 Filter 拦截链。然后反向遍历 Filter 对 Invoke 调用来进行增强。
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { Invoker<T> last = invoker; List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); if (filters.size() > 0) { for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker<T> next = last; last = new Invoker<T>() { public Class<T> 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; }
对于 Filter, InvokerListener, ExportListener 这些扩展点,可以用 @Activate 来自动激活。
@Activate // 无条件自动激活 public class XxxFilter implements Filter { // ... }
或者:
@Activate("xxx") // 当配置了xxx参数,并且参数为有效值时激活,比如配了cache="lru",自动激活CacheFilter。 public class XxxFilter implements Filter { // ... }
或者:
@Activate(group = "provider", value = "xxx") // 只对提供方激活,group可选"provider"或"consumer"
public class XxxFilter implements Filter {
// ...
}
在项目中 ,放置扩展点配置文件 META-INF/dubbo/接口全限定名
,内容为:配置名=扩展实现类全限定名
,多个实现类用换行符分隔。然后在项目启动的时候 dubbo 就会扫描所有的 jar 包内的同名文件,然后进行合并。
dubbo 里面的 Listener 与 Filter 扩展点,应用最广的还是 Filter。在 dubbo 的内部也大量的使用了 Filter 进行扩展。
同时我们可以使用使用自定义的 Filter 进行自定义的扩展,举一个最简单的例子就是日志根据。基于 dubbo 进行服务间的调用,一般都会涉及到很多个系统之间的调用。这里就可以使用 Filter 对于服务间的调用进行过滤。
黑白名单
public class AuthorityFilter implements Filter { private static final Logger LOGGER = LoggerFactory.getLogger(AuthorityFilter.class); private IpWhiteList ipWhiteList; //dubbo通过setter方式自动注入 public void setIpWhiteList(IpWhiteList ipWhiteList) { this.ipWhiteList = ipWhiteList; } @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { if (!ipWhiteList.isEnabled()) { LOGGER.debug("白名单禁用"); return invoker.invoke(invocation); } String clientIp = RpcContext.getContext().getRemoteHost(); LOGGER.debug("访问ip为{}", clientIp); List<String> allowedIps = ipWhiteList.getAllowedIps(); if (allowedIps.contains(clientIp)) { return invoker.invoke(invocation); } else { return new RpcResult(); } } }
日志
public class TradeIdFilter implements Filter { @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { Result result = null; String role = invoker.getUrl().getParameter(Constants.SIDE_KEY); if (Constants.CONSUMER.equals(role)) {// consumer String tradeId = UUID.randomUUID().toString(); RpcContext.getContext().setAttachment("tradeId", tradeId); MDC.put("tradeId", tradeId); } else if (Constants.PROVIDER.equals(role)) {// provider String tradeId = RpcContext.getContext().getAttachment("tradeId"); MDC.put("tradeId", tradeId); } try { result = invoker.invoke(invocation); } catch (Exception e) { throw new RpcException(e); } finally { MDC.clear(); } return result; } }