Dubbo源码学习--Filter过滤器

冷不防 2022-05-27 08:29 195阅读 0赞

Dubbo提供反过滤器应该类似servlet中我们经常用到的Filter，是一种递归的链式调用，用来在远程调用真正执行的前后加入一些逻辑。

Filter接口及实现类如下：

![70][]

@SPI
    public interface Filter {
    
        /**
         * do invoke filter.
         * <p>
         * <code>
         * // before filter
         * Result result = invoker.invoke(invocation);
         * // after filter
         * return result;
         * </code>
         *
         * @param invoker    service
         * @param invocation invocation.
         * @return invoke result.
         * @throws RpcException
         * @see com.alibaba.dubbo.rpc.Invoker#invoke(Invocation)
         */
        Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException;
    
    }

接下来我们简单看一个Filter的实现类AccessLogFilter，简单来说就是在invoke方法中记录一下接口调用的相关日志。

@Activate(group = Constants.PROVIDER, value = Constants.ACCESS_LOG_KEY)
    public class AccessLogFilter implements Filter {
    
        private static final Logger logger = LoggerFactory.getLogger(AccessLogFilter.class);
    
        private static final String ACCESS_LOG_KEY = "dubbo.accesslog";
    
        private static final String FILE_DATE_FORMAT = "yyyyMMdd";
    
        private static final String MESSAGE_DATE_FORMAT = "yyyy-MM-dd HH:mm:ss";
    
        private static final int LOG_MAX_BUFFER = 5000;
    
        private static final long LOG_OUTPUT_INTERVAL = 5000;
    
        private final ConcurrentMap<String, Set<String>> logQueue = new ConcurrentHashMap<String, Set<String>>();
    
        private final ScheduledExecutorService logScheduled = Executors.newScheduledThreadPool(2, new NamedThreadFactory("Dubbo-Access-Log", true));
    
        private volatile ScheduledFuture<?> logFuture = null;
    
        private void init() {
            if (logFuture == null) {
                synchronized (logScheduled) {
                    if (logFuture == null) {
                        logFuture = logScheduled.scheduleWithFixedDelay(new LogTask(), LOG_OUTPUT_INTERVAL, LOG_OUTPUT_INTERVAL, TimeUnit.MILLISECONDS);
                    }
                }
            }
        }
    
        private void log(String accesslog, String logmessage) {
            init();
            Set<String> logSet = logQueue.get(accesslog);
            if (logSet == null) {
                logQueue.putIfAbsent(accesslog, new ConcurrentHashSet<String>());
                logSet = logQueue.get(accesslog);
            }
            if (logSet.size() < LOG_MAX_BUFFER) {
                logSet.add(logmessage);
            }
        }
    
        public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
            try {
                String accesslog = invoker.getUrl().getParameter(Constants.ACCESS_LOG_KEY);
                if (ConfigUtils.isNotEmpty(accesslog)) {
                    RpcContext context = RpcContext.getContext();
                    String serviceName = invoker.getInterface().getName();
                    String version = invoker.getUrl().getParameter(Constants.VERSION_KEY);
                    String group = invoker.getUrl().getParameter(Constants.GROUP_KEY);
                    StringBuilder sn = new StringBuilder();
                    sn.append("[").append(new SimpleDateFormat(MESSAGE_DATE_FORMAT).format(new Date())).append("] ").append(context.getRemoteHost()).append(":").append(context.getRemotePort())
                            .append(" -> ").append(context.getLocalHost()).append(":").append(context.getLocalPort())
                            .append(" - ");
                    if (null != group && group.length() > 0) {
                        sn.append(group).append("/");
                    }
                    sn.append(serviceName);
                    if (null != version && version.length() > 0) {
                        sn.append(":").append(version);
                    }
                    sn.append(" ");
                    sn.append(inv.getMethodName());
                    sn.append("(");
                    Class<?>[] types = inv.getParameterTypes();
                    if (types != null && types.length > 0) {
                        boolean first = true;
                        for (Class<?> type : types) {
                            if (first) {
                                first = false;
                            } else {
                                sn.append(",");
                            }
                            sn.append(type.getName());
                        }
                    }
                    sn.append(") ");
                    Object[] args = inv.getArguments();
                    if (args != null && args.length > 0) {
                        sn.append(JSON.toJSONString(args));
                    }
                    String msg = sn.toString();
                    if (ConfigUtils.isDefault(accesslog)) {
                        LoggerFactory.getLogger(ACCESS_LOG_KEY + "." + invoker.getInterface().getName()).info(msg);
                    } else {
                        log(accesslog, msg);
                    }
                }
            } catch (Throwable t) {
                logger.warn("Exception in AcessLogFilter of service(" + invoker + " -> " + inv + ")", t);
            }
            return invoker.invoke(inv);
        }
    
        private class LogTask implements Runnable {
            public void run() {
                try {
                    if (logQueue != null && logQueue.size() > 0) {
                        for (Map.Entry<String, Set<String>> entry : logQueue.entrySet()) {
                            try {
                                String accesslog = entry.getKey();
                                Set<String> logSet = entry.getValue();
                                File file = new File(accesslog);
                                File dir = file.getParentFile();
                                if (null != dir && !dir.exists()) {
                                    dir.mkdirs();
                                }
                                if (logger.isDebugEnabled()) {
                                    logger.debug("Append log to " + accesslog);
                                }
                                if (file.exists()) {
                                    String now = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date());
                                    String last = new SimpleDateFormat(FILE_DATE_FORMAT).format(new Date(file.lastModified()));
                                    if (!now.equals(last)) {
                                        File archive = new File(file.getAbsolutePath() + "." + last);
                                        file.renameTo(archive);
                                    }
                                }
                                FileWriter writer = new FileWriter(file, true);
                                try {
                                    for (Iterator<String> iterator = logSet.iterator();
                                         iterator.hasNext();
                                         iterator.remove()) {
                                        writer.write(iterator.next());
                                        writer.write("\r\n");
                                    }
                                    writer.flush();
                                } finally {
                                    writer.close();
                                }
                            } catch (Exception e) {
                                logger.error(e.getMessage(), e);
                            }
                        }
                    }
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    
    }

接下来我们看一下所有的Filter是如果被调用的，关键代码在ProtocolFilterWrapper中，采用装饰器模式，在调用所有协议会执行ProtocolFilterWrapper这个类，这样就给我们实现Filter相关的机制提供了帮助，在ProtocolFilterWrapper中会将所有的Filter组装成一个链，当链中所有节点运行完之后才会真正执行最终的Invoker。

public class ProtocolFilterWrapper implements Protocol {
    
        private final Protocol protocol;
    
        public ProtocolFilterWrapper(Protocol protocol) {
            if (protocol == null) {
                throw new IllegalArgumentException("protocol == null");
            }
            this.protocol = protocol;
        }
    
        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;
    				//创建Filter的执行链，当所有的Filter执行完毕之后再执行invoker
                    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方法
                        public Result invoke(Invocation invocation) throws RpcException {
                            return filter.invoke(next, invocation);
                        }
    
                        public void destroy() {
                            invoker.destroy();
                        }
    
                        @Override
                        public String toString() {
                            return invoker.toString();
                        }
                    };
                }
            }
    		//返回的一个带有Filter链的invoker
            return last;
        }
    
        public int getDefaultPort() {
            return protocol.getDefaultPort();
        }
    
        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);
        }
    
        public void destroy() {
            protocol.destroy();
        }
    
    }

[70]: /images/20220527/9535e23243774bbd9b4c80b2d01f36e9.png