dubbo系列之-过滤器-2021-01-20
2021-01-20 本文已影响0人
five_year
背景
前面分析了qos wrapper，可以说是dubbo 在服务发布的过程中注入的一些额外功能，今天要介绍的过滤器Filter，主要面向在调用过程中扩展功能，虽然我们还没有介绍过dubbo服务端的调用过程，我这里简单介绍下如下图，这有助于对Filter的了解。
image
应用
上图粉红色背景的流程框图中是dubbo 默认携带的并且生效的Filter，开发人员也可以收到增加Filter，公司目前用的限流框架（Sentinel 后面分享）也是基于Filter 实现的，这块可以说是dubbo扩展中比较热门的。建议大家都手动debug去画画链路图，这块也是面试的高频，第一个红色背景流程框为netty接收到消费者请求解码之后交给dubbo的入口，最后两个红色背景流程框为Javassist动态代理类，调用具体的实现bean的入口。
过滤器介绍
生产者默认会使用到8大过滤器，EchoFilter,ClassLoaderFilter,GenericFilter,ContextFilter,
TraceFilter,TimeoutFilter,MonitorFilter,ExceptionFilter。新版本dubbo将监听处理器也放入了filter 的内部类中进行处理。
过滤器的加载
过滤器的加载是在服务export 阶段通过 ProtocolFilterWrapper 包装器进行构建的
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，比如有些过滤器只需要在提供者，并不需要在消费端生效）。然后遍历filter，将头尾通过invoker内部类联系在一起，参数传入的是 RegistryProtocol$InvokerDelegate invoker 然后包装filter链，最后返回 CallbackRegistrationInvoker invoker，这两块流程图黄色背景部分有体现。
过滤器介绍
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);
}
这个过滤器比较简单，主要配合测试服务，如果调用的方法为名称为"$echo"，就原样返回。
GenericFilter
这个过滤器要好好分析，我第一次面试的时候就被问到“泛化怎么实现？”
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行，从请求第一个参数中获取请求的方法名
代码17行从请求参数中获取generic，这个参数为泛化序列化形式，文档只介绍了true，走不认的序列化方式，其实还支持nativejava，bean，protobuf-json等
141行，包装解析之后的方法名，参数等调用
TraceFilter
介绍之前我们先来演示一番
我们telnet到dubbo 执行：trace com.shizhuang.duapp.order.api.service.core.buyer.OrderBuyerApi
image
返回了远程ip，调用的服务，参数，返回值，处理时间，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连接中发送信息
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
该过滤器主要作为监控的切入点，比如统计请求次数，请求时间，失败比例等，将信息收集之后发往用户自定义的监控消费服务，可以是一个dubbo消费者，也可以是prometheus等。
ExceptionFilter
错误拦截器会将bean中抛出的错误，进行包装到返回给消费端的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;
}
}
}
总结
还是那句话，filter 对于dubbo的扩展是常用的工具。
