前往小程序,Get更优阅读体验!
立即前往
首页
学习
活动
专区
工具
TVP
发布
社区首页 >专栏 >聊聊dubbo的Filter

聊聊dubbo的Filter

原创
作者头像
code4it
发布2019-06-17 23:16:36
1.1K0
发布2019-06-17 23:16:36
举报
文章被收录于专栏:码匠的流水账

本文主要研究一下dubbo的Filter

Filter

dubbo-2.7.2/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/Filter.java

代码语言:javascript
复制
@SPI
public interface Filter {
    /**
     * Does not need to override/implement this method.
     */
    Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException;
​
    /**
     * Filter itself should only be response for passing invocation, all callbacks has been placed into {@link Listener}
     *
     * @param appResponse
     * @param invoker
     * @param invocation
     * @return
     */
    @Deprecated
    default Result onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
        return appResponse;
    }
​
    interface Listener {
​
        void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation);
​
        void onError(Throwable t, Invoker<?> invoker, Invocation invocation);
    }
​
}
  • Filter定义了invoke、onResponse方法,另外还定义了Listener接口,该接口定义了onResponse、onError方法

ProtocolFilterWrapper

dubbo-2.7.2/dubbo-rpc/dubbo-rpc-api/src/main/java/org/apache/dubbo/rpc/protocol/ProtocolFilterWrapper.java

代码语言:javascript
复制
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;
        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 Class<T> getInterface() {
                        return invoker.getInterface();
                    }
​
                    @Override
                    public URL getUrl() {
                        return invoker.getUrl();
                    }
​
                    @Override
                    public boolean isAvailable() {
                        return invoker.isAvailable();
                    }
​
                    @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);
    }
​
    @Override
    public int getDefaultPort() {
        return protocol.getDefaultPort();
    }
​
    @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));
    }
​
    @Override
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
            return protocol.refer(type, url);
        }
        return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
    }
​
    @Override
    public void destroy() {
        protocol.destroy();
    }
​
    static class CallbackRegistrationInvoker<T> implements Invoker<T> {
​
        private final Invoker<T> filterInvoker;
        private final List<Filter> filters;
​
        public CallbackRegistrationInvoker(Invoker<T> filterInvoker, List<Filter> filters) {
            this.filterInvoker = filterInvoker;
            this.filters = filters;
        }
​
        @Override
        public Result invoke(Invocation invocation) throws RpcException {
            Result asyncResult = filterInvoker.invoke(invocation);
​
            asyncResult.thenApplyWithContext(r -> {
                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) {
                            listener.onResponse(r, filterInvoker, invocation);
                        }
                    } else {
                        filter.onResponse(r, filterInvoker, invocation);
                    }
                }
                return r;
            });
​
            return asyncResult;
        }
​
        @Override
        public Class<T> getInterface() {
            return filterInvoker.getInterface();
        }
​
        @Override
        public URL getUrl() {
            return filterInvoker.getUrl();
        }
​
        @Override
        public boolean isAvailable() {
            return filterInvoker.isAvailable();
        }
​
        @Override
        public void destroy() {
            filterInvoker.destroy();
        }
    }
}
  • ProtocolFilterWrapper实现了Protocol接口,它定义了一个静态类CallbackRegistrationInvoker,该类实现了Invoker接口,其invoke方法首先会调用filterInvoker的invoke方法获取asyncResult,之后通过thenApplyWithContext注册rpc调用完成时的回调,这里会挨个遍历filters,回调每个filter的onResponse方法

小结

  • Filter定义了invoke、onResponse方法,另外还定义了Listener接口,该接口定义了onResponse、onError方法
  • Filter定义的invoke方法返回的Result有个抽象类为AbstractResult,而AbstractResult有几个子类,分别为AppResponse、AsyncRpcResult(替代原来的RpcResult)
  • ProtocolFilterWrapper定义了一个静态类CallbackRegistrationInvoker,该类实现了Invoker接口,其invoke方法首先会调用filterInvoker的invoke方法获取asyncResult,之后通过thenApplyWithContext注册rpc调用完成时的回调,这里会挨个遍历filters,回调每个filter的onResponse方法

doc

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。

如有侵权,请联系 cloudcommunity@tencent.com 删除。

评论
登录后参与评论
0 条评论
热度
最新
推荐阅读
目录
  • Filter
  • ProtocolFilterWrapper
  • 小结
  • doc
领券
问题归档专栏文章快讯文章归档关键词归档开发者手册归档开发者手册 Section 归档