精尽 Dubbo 源码分析 —— 服务调用（三）之远程调用（HTTP）1. 概述2. AbstractProxyProtocol3. HttpProtocol

56 阅读 0 评论 37 点赞

我是靠谱客的博主心灵美钻石，最近开发中收集的这篇文章主要介绍精尽 Dubbo 源码分析 —— 服务调用（三）之远程调用（HTTP）1. 概述2. AbstractProxyProtocol3. HttpProtocol，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

1. 概述

基于 HTTP 表单的远程调用协议，采用 Spring 的 HttpInvoker 实现

涉及类图（红圈部分）如下：
在这里插入图片描述

2. AbstractProxyProtocol

现 AbstractProtocol 抽象类，Proxy 协议抽象类。为 HttpProtocol 、RestProtocol 等子类，提供公用的服务暴露、服务引用的公用方法，同时定义了如下抽象方法，用于不同子类协议实现类的自定义的逻辑：

/**
     * 获得异常对应的错误码
     *
     * @param e 异常
     * @return 错误码
     */
    protected int getErrorCode(Throwable e) {
        return RpcException.UNKNOWN_EXCEPTION;
    }

    /**
     * 执行暴露，并返回取消暴露的回调 Runnable
     *
     * @param impl 服务 Proxy 对象
     * @param type 服务接口
     * @param url URL
     * @param <T> 服务接口
     * @return 消暴露的回调 Runnable
     * @throws RpcException 当发生异常
     */
    protected abstract <T> Runnable doExport(T impl, Class<T> type, URL url) throws RpcException;

    /**
     * 执行引用，并返回调用远程服务的 Service 对象
     *
     * @param type 服务接口
     * @param url URL
     * @param <T> 服务接口
     * @return 调用远程服务的 Service 对象
     * @throws RpcException 当发生异常
     */
    protected abstract <T> T doRefer(Class<T> type, URL url) throws RpcException;

构造方法：

/**
     * 需要抛出的异常类集合，详见 {@link #refer(Class, URL)} 方法。
     */
    private final List<Class<?>> rpcExceptions = new CopyOnWriteArrayList<Class<?>>();
    /**
     * ProxyFactory 对象
     */
    private ProxyFactory proxyFactory;

    public AbstractProxyProtocol() {
    }

    public AbstractProxyProtocol(Class<?>... exceptions) {
        for (Class<?> exception : exceptions) {
            addRpcException(exception);
        }
    }
    public void addRpcException(Class<?> exception) {
    this.rpcExceptions.add(exception);
}
}

2.1 export

    public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
        // 获得服务键
        final String uri = serviceKey(invoker.getUrl());
        // 获得 Exporter 对象。若已经暴露，直接返回。
        Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
        if (exporter != null) {
            return exporter;
        }
        // 执行暴露服务
        final Runnable runnable = doExport(proxyFactory.getProxy(invoker), invoker.getInterface(), invoker.getUrl());
        // 创建 Exporter 对象
        exporter = new AbstractExporter<T>(invoker) {

            @Override
            public void unexport() {
                // 取消暴露
                super.unexport();
                exporterMap.remove(uri);
                // 执行取消暴露的回调
                if (runnable != null) {
                    try {
                        runnable.run();
                    } catch (Throwable t) {
                        logger.warn(t.getMessage(), t);
                    }
                }
            }

        };
        // 添加到 Exporter 集合
        exporterMap.put(uri, exporter);
        return exporter;
    }

2.2 refer

    public <T> Invoker<T> refer(final Class<T> type, final URL url) throws RpcException {
        // 执行引用服务
        final Invoker<T> target = proxyFactory.getInvoker(doRefer(type, url), type, url);
        // 创建 Invoker 对象
        Invoker<T> invoker = new AbstractInvoker<T>(type, url) {

            @Override
            protected Result doInvoke(Invocation invocation) throws Throwable {
                try {
                    // 调用
                    Result result = target.invoke(invocation);
                    // 若返回结果带有异常，并且需要抛出，则抛出异常。
                    Throwable e = result.getException();
                    if (e != null) {
                        for (Class<?> rpcException : rpcExceptions) {
                            if (rpcException.isAssignableFrom(e.getClass())) {
                                throw getRpcException(type, url, invocation, e);
                            }
                        }
                    }
                    return result;
                } catch (RpcException e) {
                    // 若是未知异常，获得异常对应的错误码
                    if (e.getCode() == RpcException.UNKNOWN_EXCEPTION) {
                        e.setCode(getErrorCode(e.getCause()));
                    }
                    throw e;
                } catch (Throwable e) {
                    // 抛出 RpcException 异常
                    throw getRpcException(type, url, invocation, e);
                }
            }

        };
        // 添加到 Invoker 集合。
        invokers.add(invoker);
        return invoker;
    }

3. HttpProtocol

现 AbstractProxyProtocol 抽象类，dubbo:// 协议实现类。

/**
 * HttpProtocol
 *
 * HTTP 协议实现类
 */
public class HttpProtocol extends AbstractProxyProtocol {

    /**
     * 默认服务器端口
     */
    public static final int DEFAULT_PORT = 80;
    /**
     * Http 服务器集合
     *
     * key：ip:port
     */
    private final Map<String, HttpServer> serverMap = new ConcurrentHashMap<String, HttpServer>();
    /**
     * Spring HttpInvokerServiceExporter 集合
     *
     * key：path 服务名
     */
    private final Map<String, HttpInvokerServiceExporter> skeletonMap = new ConcurrentHashMap<String, HttpInvokerServiceExporter>();
    /**
     * HttpBinder$Adaptive 对象
     */
    private HttpBinder httpBinder;

    public HttpProtocol() {
        super(RemoteAccessException.class);
    }

    public void setHttpBinder(HttpBinder httpBinder) {
        this.httpBinder = httpBinder;
    }

    public int getDefaultPort() {
        return DEFAULT_PORT;
    }

3.1 InternalHandler

private class InternalHandler implements HttpHandler {

    @Override
    public void handle(HttpServletRequest request, HttpServletResponse response) throws ServletException {
        String uri = request.getRequestURI();
        // 获得 HttpInvokerServiceExporter 对象
        HttpInvokerServiceExporter skeleton = skeletonMap.get(uri);
        // 必须是 POST 请求
        if (!request.getMethod().equalsIgnoreCase("POST")) {
            response.setStatus(500);
        // 执行调用
        } else {
            RpcContext.getContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
            try {
                skeleton.handleRequest(request, response);
            } catch (Throwable e) {
                throw new ServletException(e);
            }
        }
    }

}

3.2 doRefer

    protected <T> T doRefer(final Class<T> serviceType, final URL url) throws RpcException {
        // 创建 HttpInvokerProxyFactoryBean 对象
        final HttpInvokerProxyFactoryBean httpProxyFactoryBean = new HttpInvokerProxyFactoryBean();
        httpProxyFactoryBean.setServiceUrl(url.toIdentityString());
        httpProxyFactoryBean.setServiceInterface(serviceType);
        // 创建执行器 SimpleHttpInvokerRequestExecutor 对象
        String client = url.getParameter(Constants.CLIENT_KEY);
        if (client == null || client.length() == 0 || "simple".equals(client)) {
            SimpleHttpInvokerRequestExecutor httpInvokerRequestExecutor = new SimpleHttpInvokerRequestExecutor() {
                protected void prepareConnection(HttpURLConnection con,
                                                 int contentLength) throws IOException {
                    super.prepareConnection(con, contentLength);
                    con.setReadTimeout(url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
                    con.setConnectTimeout(url.getParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT));
                }
            };
            httpProxyFactoryBean.setHttpInvokerRequestExecutor(httpInvokerRequestExecutor);
        // 创建执行器 HttpComponentsHttpInvokerRequestExecutor 对象
        } else if ("commons".equals(client)) {
            HttpComponentsHttpInvokerRequestExecutor httpInvokerRequestExecutor = new HttpComponentsHttpInvokerRequestExecutor();
            httpInvokerRequestExecutor.setReadTimeout(url.getParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT));
            httpProxyFactoryBean.setHttpInvokerRequestExecutor(httpInvokerRequestExecutor);
        } else {
            throw new IllegalStateException("Unsupported http protocol client " + client + ", only supported: simple, commons");
        }
        httpProxyFactoryBean.afterPropertiesSet();
        // 返回 HttpInvokerProxyFactoryBean 对象
        return (T) httpProxyFactoryBean.getObject();
    }