dubbo中的Filter链原理及应用

转载：https://www.jianshu.com/p/f390bb88574d

filter在dubbo中的应用非常广泛，它可以对服务端、消费端的调用过程进行拦截，从而对dubbo进行功能上的扩展，我们所熟知的RpcContext就用到了filter。本文主要尝试从以下3个方面来简单介绍一下dubbo中的filter：
1.filter链原理
2.自定义filter
3.使用filter透传traceId

1.filter链原理

dubbo中filter链的入口在ProtocolFilterWrapper中，这是Protocol的一个包装类，在其服务暴露和服务引用时都进行了构建filter链的工作。

// 构建filter链 private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { Invoker<T> last = invoker; // 获取可用的filter列表 List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); if (!filters.isEmpty()) { for (int i = filters.size() - 1; i >= 0; i--) { final Filter filter = filters.get(i); final Invoker<T> next = last; // 典型的装饰器模式，将invoker用filter逐层进行包装 last = new Invoker<T>() { public Class<T> getInterface() { return invoker.getInterface(); } public URL getUrl() { return invoker.getUrl(); } public boolean isAvailable() { return invoker.isAvailable(); } // 重点，每个filter在执行invoke方法时，会触发其下级节点的invoke方法，最后一级节点即为最原始的服务 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; } // 服务端暴露服务 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)); } // 客户端引用服务 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节点都为原始的invoker服务增加了功能，是典型的装饰器模式。构建filter链的核心在于filter列表的获取，也就是这一行代码：

List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group); 

通过Filter的ExtendLoader实例获取其激活的filter列表，getActivateExtension逻辑分为两部分：
1.加载标注了Activate注解的filter列表
2.加载用户在spring配置文件中手动注入的filter列表

public List<T> getActivateExtension(URL url, String key, String group) { // 根据key来获取服务方/消费方自定义的filter列表 String value = url.getParameter(key); return getActivateExtension(url, value == null || value.length() == 0 ? null : Constants.COMMA_SPLIT_PATTERN.split(value), group); } public List<T> getActivateExtension(URL url, String[] values, String group) { List<T> exts = new ArrayList<T>(); List<String> names = values == null ? new ArrayList<String>(0) : Arrays.asList(values); // 如果用户配置的filter列表名称中不包含-default，则加载标注了Activate注解的filter列表 if (!names.contains(Constants.REMOVE_VALUE_PREFIX + Constants.DEFAULT_KEY)) { // 加载配置文件，获取所有标注有Activate注解的类，存入cachedActivates中 getExtensionClasses(); for (Map.Entry<String, Activate> entry : cachedActivates.entrySet()) { String name = entry.getKey(); Activate activate = entry.getValue(); // Activate注解可以指定group，这里是看注解指定的group与我们要求的group是否匹配 if (isMatchGroup(group, activate.group())) { T ext = getExtension(name); // 对于每一个dubbo中原生的filter，需要满足以下3个条件才会被加载： // 1.用户配置的filter列表中不包含该名称的filter // 2.用户配置的filter列表中不包含该名称前加了"-"的filter // 3.该Activate注解被激活，具体激活条件随后详解 if (!names.contains(name) && !names.contains(Constants.REMOVE_VALUE_PREFIX + name) && isActive(activate, url)) { exts.add(ext); } } } // 对加载的dubbo原生的filter列表进行排序，ActivateComparator排序器会根据Activate注解的before、after、order属性对filter列表排序 Collections.sort(exts, ActivateComparator.COMPARATOR); } // 加载用户在spring配置文件中配置的filter列表 List<T> usrs = new ArrayList<T>(); for (int i = 0; i < names.size(); i++) { String name = names.get(i); // 针对用户配置的每一个filter，需要满足以下两个条件才会被加载： // 1.名称不是以"-"开头 // 2.用户配置的所有filter列表中不包含-name的filter if (!name.startsWith(Constants.REMOVE_VALUE_PREFIX) && !names.contains(Constants.REMOVE_VALUE_PREFIX + name)) { // 用户自己配置filter列表时，可以使用default的key来代表dubbo原生的filter列表，这样一来就可以控制dubbo原生filter列表和用户自定义filter列表之间的相对顺序 if (Constants.DEFAULT_KEY.equals(name)) { if (!usrs.isEmpty()) { exts.addAll(0, usrs); usrs.clear(); } } else { T ext = getExtension(name); usrs.add(ext); } } } if (!usrs.isEmpty()) { exts.addAll(usrs); } return exts; } 

判断Activate注解是否被激活的逻辑是这样的：

private boolean isActive(Activate activate, URL url) { // 如果注解没有配置value属性，则一定是激活的 String[] keys = activate.value(); if (keys == null || keys.length == 0) { return true; } // 对配置了value属性的注解，如果服务的url属性中存在与value属性值相匹配的属性且改属性值不为空，则该注解也是激活的 for (String key : keys) { for (Map.Entry<String, String> entry : url.getParameters().entrySet()) { String k = entry.getKey(); String v = entry.getValue(); if ((k.equals(key) || k.endsWith("." + key)) && ConfigUtils.isNotEmpty(v)) { return true; } } } return false; } 

ActivateComparator比较器的规则如下，总结起来有这么几条规则：
1.before指定的过滤器，该过滤器将在这些指定的过滤器之前执行
2.after指定的过滤器，该过滤器将在这些指定的过滤器之后执行
3.order数值越小，越先执行
4.order数值相等的条件下，顺序将依赖于两个filter的加载顺序
5.before/after的优先级高于order

public class ActivateComparator implements Comparator<Object> { public static final Comparator<Object> COMPARATOR = new ActivateComparator(); public int compare(Object o1, Object o2) { if (o1 == null && o2 == null) { return 0; } if (o1 == null) { return -1; } if (o2 == null) { return 1; } if (o1.equals(o2)) { return 0; } // 配置了before/after属性时，按照规则1、2进行排序，比较完直接返回，此时指定的order值将被忽略 Activate a1 = o1.getClass().getAnnotation(Activate.class); Activate a2 = o2.getClass().getAnnotation(Activate.class); if ((a1.before().length > 0 || a1.after().length > 0 || a2.before().length > 0 || a2.after().length > 0) && o1.getClass().getInterfaces().length > 0 && o1.getClass().getInterfaces()[0].isAnnotationPresent(SPI.class)) { ExtensionLoader<?> extensionLoader = ExtensionLoader.getExtensionLoader(o1.getClass().getInterfaces()[0]); if (a1.before().length > 0 || a1.after().length > 0) { String n2 = extensionLoader.getExtensionName(o2.getClass()); for (String before : a1.before()) { if (before.equals(n2)) { return -1; } } for (String after : a1.after()) { if (after.equals(n2)) { return 1; } } } if (a2.before().length > 0 || a2.after().length > 0) { String n1 = extensionLoader.getExtensionName(o1.getClass()); for (String before : a2.before()) { if (before.equals(n1)) { return 1; } } for (String after : a2.after()) { if (after.equals(n1)) { return -1; } } } } // 没有配置before/after的条件下，按照规则3、4进行排序 int n1 = a1 == null ? 0 : a1.order(); int n2 = a2 == null ? 0 : a2.order(); // never return 0 even if n1 equals n2, otherwise, o1 and o2 will override each other in collection like HashSet return n1 > n2 ? 1 : -1; } } 

分析上面的分析，可以发现dubbo在构建filter链时非常灵活，有几个关键点在这里做一下总结：

• filter被分为两类，一类是标注了Activate注解的filter，包括dubbo原生的和用户自定义的；一类是用户在spring配置文件中手动注入的filter
• 对标注了Activate注解的filter，可以通过before、after和order属性来控制它们之间的相对顺序，还可以通过group来区分服务端和消费端
• 手动注入filter时，可以用default来代表所有标注了Activate注解的filter，以此来控制两类filter之间的顺序
• 手动注入filter时，可以在filter名称前加一个"-"表示排除某一个filter，比如说如果配置了一个-default的filter，将不再加载所有标注了Activate注解的filter

2.自定义filter

自定义filter非常简单，只需要实现Filter接口即可，对于Filter的某一个具体实现，有两种方式可以在构建filter链时将其包含进去，但无论哪种方式，都需要在Filter对应的SPI文件中进行相应的配置
1.通过标注Activate注解来实现

@Activate(group = Constants.PROVIDER) public class ProviderFilter implements Filter { @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { System.out.println("=== provider ==="); return invoker.invoke(invocation); } } 

2.在spring配置文件中通过配置filter属性来实现

<dubbo:service interface="com.alibaba.dubbo.demo.TraceIdService" ref="traceIdService" filter="providerFilter"/> 

这两种方式除了该filter在filter链中的顺序不同外，其它地方都是等价的。当然，按照上面的分析，顺序也是可以按照我们的要求来灵活控制的。

3.利用filter实现traceId透传

在微服务场景下，一次调用过程常常会涉及多个应用，在定位问题时，往往需要在多个应用中查看某一次调用链路上的日志，为了达到这个目的，一种常见的做法是在调用入口处生成一个traceId，并基于RpcContext来实现traceId的透传。

在开始进一步的尝试之前，我们不妨先来看看两个filter，大致了解下RpcContext是怎么实现traceId透传的。
客户端的ConsumerContextFilter：

@Activate(group = Constants.CONSUMER, order = -10000) public class ConsumerContextFilter implements Filter { public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { RpcContext.getContext() .setInvoker(invoker) .setInvocation(invocation) .setLocalAddress(NetUtils.getLocalHost(), 0) .setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort()); if (invocation instanceof RpcInvocation) { ((RpcInvocation) invocation).setInvoker(invoker); } try { return invoker.invoke(invocation); } finally { RpcContext.getContext().clearAttachments(); } } } 

服务端的ContextFilter：

@Activate(group = Constants.PROVIDER, order = -10000) public class ContextFilter implements Filter { public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { Map<String, String> attachments = invocation.getAttachments(); if (attachments != null) { attachments = new HashMap<String, String>(attachments); attachments.remove(Constants.PATH_KEY); attachments.remove(Constants.GROUP_KEY); attachments.remove(Constants.VERSION_KEY); attachments.remove(Constants.DUBBO_VERSION_KEY); attachments.remove(Constants.TOKEN_KEY); attachments.remove(Constants.TIMEOUT_KEY); attachments.remove(Constants.ASYNC_KEY);// Remove async property to avoid being passed to the following invoke chain. } RpcContext.getContext() .setInvoker(invoker) .setInvocation(invocation) // .setAttachments(attachments) // merged from dubbox .setLocalAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort()); // mreged from dubbox // we may already added some attachments into RpcContext before this filter (e.g. in rest protocol) if (attachments != null) { if (RpcContext.getContext().getAttachments() != null) { RpcContext.getContext().getAttachments().putAll(attachments); } else { RpcContext.getContext().setAttachments(attachments); } } if (invocation instanceof RpcInvocation) { ((RpcInvocation) invocation).setInvoker(invoker); } try { return invoker.invoke(invocation); } finally { RpcContext.removeContext(); } } } 

通过这两个filter不难发现，之所以利用RpcContext可以实现traceId的透传，是因为invocation的存在，客户端在调用invoke方法的时候，会将当前调用的参数载体invocation透传给服务端，而服务端会从其中取出attachments属性进行相关处理后在重新设置到invocation中向后传递，因此只需要在客户端将traceId设置到attachments中即可。

于是我们开始以下尝试：
服务端接口：

public interface TraceIdService { void test(String key); } 

服务端实现：

public class TraceIdServiceImpl implements TraceIdService { @Override public void test(String key) { String traceId = RpcContext.getContext().getAttachment("traceId"); System.out.println("key = " + key + ", traceId = " + traceId); } } 

客户端代码：

public class TraceIdConsumer { public static void main(String[] args) { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"}); context.start(); TraceIdService traceIdService = (TraceIdService) context.getBean("traceIdService"); // get remote service proxy RpcContext.getContext().setAttachment("traceId", String.valueOf(System.currentTimeMillis())); System.out.println(RpcContext.getContext().getAttachments()); traceIdService.test("1"); System.out.println(RpcContext.getContext().getAttachments()); traceIdService.test("2"); } } 

以上代码的执行结果如下：

客户端输出：
{traceId=1538746615202} {} 服务端输出： key = 1, traceId = 1538746615202 key = 2, traceId = null 

我们发现，在第一次调用中，traceId确实从客户端透传到了服务端，但是在第二次调用时神奇的消失了！而这正是filter捣的鬼。在ConsumerContextFilter的finally子句中，我们发现attachments对象被清空了，而在服务端ContextFilter中，整个context对象都被清空了！！！

为了解决这个问题，我们需要在每次调用前都重新设置下attachments对象，也就是在客户端给调用链新增一个设置attachments对象的功能。前面我们说过，dubbo中每一个filter节点都为原始的invoker服务增加了功能，是典型的装饰器模式。看到这里你想到了什么？是的，没错。我们可以新增一个filter来完成这一功能。

@Activate(group = Constants.CONSUMER) public class TraceIdFilter implements Filter { @Override public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException { String traceId = String.valueOf(System.currentTimeMillis()); RpcContext.getContext().setAttachment("traceId", traceId); System.out.println("traceId = " + traceId); return invoker.invoke(invocation); } } 

此时在客户端中注释小设置attachments的代码，再次执行代码的输出如下，此时两次调用，traceId都可以正确地从客户端传递到服务端，完美؏؏☝ᖗ乛◡乛ᖘ☝؏؏

客户端输出：
traceId = 1538749616953
traceId = 1538749617199

服务端输出：
key = 1, traceId = 1538749616953
key = 2, traceId = 1538749617199

作者：shysheng
链接：https://www.jianshu.com/p/f390bb88574d
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。