背景
前面分析了qos wrapper，可以說是dubbo 在服務(wù)發(fā)布的過程中注入的一些額外功能，今天要介紹的過濾器Filter，主要面向在調(diào)用過程中擴(kuò)展功能，雖然我們還沒有介紹過dubbo服務(wù)端的調(diào)用過程，我這里簡單介紹下如下圖，這有助于對(duì)Filter的了解。
應(yīng)用
上圖粉紅色背景的流程框圖中是dubbo 默認(rèn)攜帶的并且生效的Filter，開發(fā)人員也可以收到增加Filter，公司目前用的限流框架（Sentinel 后面分享）也是基于Filter 實(shí)現(xiàn)的，這塊可以說是dubbo擴(kuò)展中比較熱門的。建議大家都手動(dòng)debug去畫畫鏈路圖，這塊也是面試的高頻，第一個(gè)紅色背景流程框?yàn)閚etty接收到消費(fèi)者請(qǐng)求解碼之后交給dubbo的入口，最后兩個(gè)紅色背景流程框?yàn)镴avassist動(dòng)態(tài)代理類，調(diào)用具體的實(shí)現(xiàn)bean的入口。
過濾器介紹
生產(chǎn)者默認(rèn)會(huì)使用到8大過濾器，EchoFilter,ClassLoaderFilter,GenericFilter,ContextFilter,
TraceFilter,TimeoutFilter,MonitorFilter,ExceptionFilter。新版本dubbo將監(jiān)聽處理器也放入了filter 的內(nèi)部類中進(jìn)行處理。
過濾器的加載
過濾器的加載是在服務(wù)export 階段通過 ProtocolFilterWrapper 包裝器進(jìn)行構(gòu)建的
private static Invoker<T> buildInvokerChain(Invoker<T> invoker, key, group) {
Invoker<T> last = invoker;
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;
last = new Invoker<T>() {
//....
@Override
public Result invoke(Invocation invocation) throws RpcException {
Result asyncResult;
try {
asyncResult = filter.invoke(next, invocation);
} catch (Exception e) {
// onError callback
}
return asyncResult;
}
};
}
}
return new CallbackRegistrationInvoker<>(last, filters);
}
@Override
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));
}
代碼第3行通過spi加載所有滿足條件的filter，并排好序（過濾條件為key和group，比如有些過濾器只需要在提供者，并不需要在消費(fèi)端生效）。然后遍歷filter，將頭尾通過invoker內(nèi)部類聯(lián)系在一起，參數(shù)傳入的是 RegistryProtocol$InvokerDelegate invoker 然后包裝filter鏈，最后返回 CallbackRegistrationInvoker invoker，這兩塊流程圖黃色背景部分有體現(xiàn)。
過濾器介紹
EchoFilter
public class EchoFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
if (inv.getMethodName().equals("$echo") && inv.getArguments() != null
&& inv.getArguments().length == 1) {
return AsyncRpcResult.newDefaultAsyncResult(inv.getArguments()[0], inv);
}
return invoker.invoke(inv);
}
這個(gè)過濾器比較簡單，主要配合測(cè)試服務(wù)，如果調(diào)用的方法為名稱為"$echo"，就原樣返回。
GenericFilter
這個(gè)過濾器要好好分析，我第一次面試的時(shí)候就被問到“泛化怎么實(shí)現(xiàn)？”
public class GenericFilter extends ListenableFilter {
@Override
public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
if ((inv.getMethodName().equals("$invoke") || inv.getMethodName().equals($INVOKE_ASYNC))
&& inv.getArguments() != null
&& inv.getArguments().length == 3
&& !GenericService.class.isAssignableFrom(invoker.getInterface())) {
String name = ((String) inv.getArguments()[0]).trim();
String[] types = (String[]) inv.getArguments()[1];
Object[] args = (Object[]) inv.getArguments()[2];
try {
Method method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), name, types);
Class<?>[] params = method.getParameterTypes();
if (args == null) {
args = new Object[params.length];
}
String generic = inv.getAttachment(GENERIC_KEY);
if (StringUtils.isBlank(generic)) {
generic = RpcContext.getContext().getAttachment(GENERIC_KEY);
}
if (StringUtils.isEmpty(generic)
|| ProtocolUtils.isDefaultGenericSerialization(generic)) {
args = PojoUtils.realize(args, params, method.getGenericParameterTypes());
} else if (ProtocolUtils.isJavaGenericSerialization(generic)) {
for (int i = 0; i < args.length; i++) {
if (byte[].class == args[i].getClass()) {
try (UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream((byte[]) args[i])) {
args[i] = ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension(GENERIC_SERIALIZATION_NATIVE_JAVA)
.deserialize(null, is).readObject();
} catch (Exception e) {}
} else {
throw new RpcException();
}
}
} else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
for (int i = 0; i < args.length; i++) {
if (args[i] instanceof JavaBeanDescriptor) {
args[i] = JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) args[I]);
} else {
throw new RpcException();
}
}
} else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
// as proto3 only accept one protobuf parameter
if (args.length == 1 && args[0] instanceof String) {
try (UnsafeByteArrayInputStream is =
new UnsafeByteArrayInputStream(((String) args[0]).getBytes())) {
args[0] = ExtensionLoader.getExtensionLoader(Serialization.class)
.getExtension("" + GENERIC_SERIALIZATION_PROTOBUF)
.deserialize(null, is).readObject(method.getParameterTypes()[0]);
} catch (Exception e) {}
} else {
throw new RpcException();
}
}
return invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
} catch (Exception e) {
throw new RpcException(e.getMessage(), e);
}
}
return invoker.invoke(inv);
}
代碼第8行，從請(qǐng)求第一個(gè)參數(shù)中獲取請(qǐng)求的方法名
代碼17行從請(qǐng)求參數(shù)中獲取generic，這個(gè)參數(shù)為泛化序列化形式，文檔只介紹了true，走不認(rèn)的序列化方式，其實(shí)還支持nativejava，bean，protobuf-json等
141行，包裝解析之后的方法名，參數(shù)等調(diào)用
TraceFilter
介紹之前我們先來演示一番
我們telnet到dubbo 執(zhí)行：trace com.shizhuang.duapp.order.api.service.core.buyer.OrderBuyerApi
返回了遠(yuǎn)程ip，調(diào)用的服務(wù)，參數(shù)，返回值，處理時(shí)間，trace的功能和簡單就是跟蹤下接口。
public class TraceFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
long start = System.currentTimeMillis();
Result result = invoker.invoke(invocation);
long end = System.currentTimeMillis();
if (tracers.size() > 0) {
String key = invoker.getInterface().getName() + "." + invocation.getMethodName();
Set<Channel> channels = tracers.get(key);
if (channels == null || channels.isEmpty()) {
key = invoker.getInterface().getName();
channels = tracers.get(key);
}
//獲取trace 的所有netty連接
if (CollectionUtils.isNotEmpty(channels)) {
for (Channel channel : new ArrayList<>(channels)) {
if (channel.isConnected()) {
int max = 1;
Integer m = (Integer) channel.getAttribute(TRACE_MAX);
if (m != null) { max = m; }
int count = 0;
AtomicInteger c = (AtomicInteger) channel.getAttribute(TRACE_COUNT);
count = c.getAndIncrement();
if (count < max) {
//往netty連接中發(fā)送信息
String prompt = channel.getUrl().getParameter(Constants.PROMPT_KEY, Constants.DEFAULT_PROMPT);
channel.send("\r\n" + RpcContext.getContext().getRemoteAddress() + " -> "
+ invoker.getInterface().getName()
+ "." + invocation.getMethodName()
+ "(" + JSON.toJSONString(invocation.getArguments()) + ")" + " -> " + JSON.toJSONString(result.getValue())
+ "\r\nelapsed: " + (end - start) + " ms."
+ "\r\n\r\n" + prompt);
}
if (count >= max - 1) {
channels.remove(channel);
}
} else {
channels.remove(channel);
}
}
}
}
return result;
}
MonitorFilter
該過濾器主要作為監(jiān)控的切入點(diǎn)，比如統(tǒng)計(jì)請(qǐng)求次數(shù)，請(qǐng)求時(shí)間，失敗比例等，將信息收集之后發(fā)往用戶自定義的監(jiān)控消費(fèi)服務(wù)，可以是一個(gè)dubbo消費(fèi)者，也可以是prometheus等。
ExceptionFilter
錯(cuò)誤攔截器會(huì)將bean中拋出的錯(cuò)誤，進(jìn)行包裝到返回給消費(fèi)端的result中。
static class ExceptionListener implements Listener {
@Override
public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
if (appResponse.hasException() && GenericService.class != invoker.getInterface()) {
try {
Throwable exception = appResponse.getException();
// otherwise, wrap with RuntimeException and throw back to the client
appResponse.setException(new RuntimeException(StringUtils.toString(exception)));
return;
} catch (Throwable e) {
logger.warn("Fail to ExceptionFilter when called by " + RpcContext.getContext().getRemoteHost() + ". service: " + invoker.getInterface().getName() + ", method: " + invocation.getMethodName() + ", exception: " + e.getClass().getName() + ": " + e.getMessage(), e);
return;
}
}
}
總結(jié)
還是那句話，filter 對(duì)于dubbo的擴(kuò)展是常用的工具。
