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
更多技术文章，咱们公众号见，我在公众号里等你~
image.png
