1.dubbo调用过程
以dubbo官方demo为例，在provider端，从netty接收到消息，递交给业务线程池处理开始，到真正调用到业务方法sayHello()结束，中间经过了十几个Filter的处理。见下图
屏幕快照 2019-08-04 下午1.04.21.png
那么这些Filter是如何初始化的，调用的时候又是如何执行的呢？接下来一步一步介绍。
2.dubbo服务导出过程
dubbo在进行服务导出时主要做了如下一些工作
屏幕快照 2019-08-04 下午12.55.24.png
可以看到：第二步中核心工作就包括Filter的初始化。见下ProtocolFilterWrapper#export方法
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
return protocol.export(buildInvokerChain(invoker, SERVICE_FILTER_KEY, CommonConstants.PROVIDER));
}
其中有一步是buildInvokerChain，从名字上也可以看出，这是初始化一个责任链，对应设计模式中的责任链模式。接着看这个责任链是怎么初始化的（ProtocolFilterWrapper#buildInvokerChain）
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
Invoker<T> last = invoker;
//利用dubbo spi得到实现了Filter接口的所有实例，形成List<Filter>
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
if (!filters.isEmpty()) {
//遍历这个List<Filter>，形成一个Invoker链表
for (int i = filters.size() - 1; i >= 0; i--) {
final Filter filter = filters.get(i);
final Invoker<T> next = last;
last = new Invoker<T>() {
@Override
public Class<T> getInterface() {
return invoker.getInterface();
}
@Override
public URL getUrl() {
return invoker.getUrl();
}
@Override
public boolean isAvailable() {
return invoker.isAvailable();
}
//执行责任链上某个节点的invoke逻辑时，同时会传入next节点信息，以支持链式执行
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult;
try {
asyncResult = filter.invoke(next, invocation);
} catch (Exception e) {
// onError callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
listener.onError(e, invoker, invocation);
}
}
throw e;
}
return asyncResult;
}
@Override
public void destroy() {
invoker.destroy();
}
@Override
public String toString() {
return invoker.toString();
}
};
}
}
return new CallbackRegistrationInvoker<>(last, filters);
}
可以看到这个Filter的责任链初始化过程
1.通过dubbo spi机制，取得所有Filter实例形成一个ArrayList<Filter>
2.遍历这个ArrayList，以next指针初始化一个Invoker的链表InvokerList
3.Invoker执行逻辑即执行Filter的invoke方法的同时，将next指针作为参数传入，以支持链式调用
整个过程结束之后，会有两个List，一个ArrayList<Filter>，一个以NEXT指针形成的InvokerList，这两个List就是dubbo Filter机制的基础。
3.Filter接口
@SPI
public interface Filter {
/**
* Does not need to override/implement this method.
*/
Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException;
interface Listener {
void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation);
void onError(Throwable t, Invoker<?> invoker, Invocation invocation);
}
}
Filter接口提供了一个invoke方法，另一个是Listener接口，有onResponse，onError两个方法。
这两部分对应着Filter对请求和响应的处理逻辑
请求的处理以AccessLogFilter#invoke的实现为例,可以看到，其首先进行AccessLog的处理，然后调用
invoker.invoke().这个invoker即之前Filter初始化的时候，以Next指针形成的那个InvokerList链表中的节点。这样一来，整个链表中的节点都会得到顺序执行。
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
try {
String accessLogKey = invoker.getUrl().getParameter(ACCESS_LOG_KEY);
if (ConfigUtils.isNotEmpty(accessLogKey)) {
AccessLogData logData = buildAccessLogData(invoker, inv);
log(accessLogKey, logData);
}
} catch (Throwable t) {
logger.warn("Exception in AccessLogFilter of service(" + invoker + " -> " + inv + ")", t);
}
return invoker.invoke(inv);
}
响应的处理可以见CallbackRegistrationInvoker#invoke
asyncResult = asyncResult.whenCompleteWithContext((r, t) -> {
for (int i = filters.size() - 1; i >= 0; i--) {
Filter filter = filters.get(i);
// onResponse callback
if (filter instanceof ListenableFilter) {
Filter.Listener listener = ((ListenableFilter) filter).listener();
if (listener != null) {
if (t == null) {
listener.onResponse(r, filterInvoker, invocation);
} else {
listener.onError(t, filterInvoker, invocation);
}
}
} else {
filter.onResponse(r, filterInvoker, invocation);
}
}
});
之前Filter初始化的时候，会形成两个list，next指针形成的那个链表用于对请求的处理，另一个ArrayList<Filter> 就是在此时执行,拿到结果之后，遍历这个ArrayList，执行其onResponse或者onError方法，如此一来，请求和响应应就会经过所有生效的Filter处理。
总结：dubbo的Filter机制是一种典型的责任链模式，这个链的基础即上述两个list。如果我们在自己的业务场景中，需要对请求或者响应做一些通用的处理，那么也很简单，直接基于dubbo spi（ https://www.jianshu.com/p/d4d7ebc8f7bb
），自定义逻辑实现Filter接口的相应方法即可。
网友评论