dubbo源码分析之过滤器Filter-12

Posted 2021-03-24 梦想空间

https://blog.csdn.net/luoyang_java/article/details/86682668

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，可以看到优秀的框架里面有很多思想都是相通的。

1、dubbo 的领域模型
在 dubbo 中主要包含以下三个核心领域模型：

Protocol 是服务域，它是 Invoker 暴露和引用的主功能入口，它负责 Invoker 的生命周期管理。
Invoker 是实体域，它是 Dubbo 的核心模型，其它模型都向它靠扰，或转换成它，它代表一个可执行体，可向它发起 invoke 调用，它有可能是一个本地的实现，也可能是一个远程的实现，也可能一个集群实现。
Invocation 是会话域，它持有调用过程中的变量，比如方法名，参数等。
dubbo 通过 Protocol 服务暴露 Invoke 这个服务调用的执行体，然后通过 Protocol 引用 Invoke 调用 Invocation 提供调用过程中的变量就可以完成一次服务的发布与调用过程。
 

2、dubbo 扩展之 Listener
dubbo 的服务发布与引用都提供了Listener 分别是 ExporterListener(服务暴露监听) 与 InvokerListener(服务调用监听)。

2.1 ExporterListener
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);

}
exported() ：在 ListenerExporterWrapper 对象进行初始化的时候就会进行调用
unexported() ：Exporter#unexport 的时候就会进行调用
2.2 InvokerListener
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);

}
referred() ：在 ListenerInvokerWrapper 对象进行初始化的时候就会进行调用
destroyed() ：Invoker#destroyed 的时候就会进行调用
3、dubbo 扩展之 Filter
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;
}
4、扩展点自动激活
对于 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 包内的同名文件，然后进行合并。

5、扩展点应用
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;
}
}