RPC架构设计及在dubbo中的实现分析

什么是RPC

RPC（Remote Procedure Call）是远程过程调用的简写，所谓远程过程调用是指应用通过网络像调用本地方法一样同步调用运行在其他计算机进程中的方法，也就是RPC是一种进程间通信的方式，RPC架构采用C/S默认架构，调用方就是客户端，服务提供方就是服务端。主要应用在分布式系统服务调用，分布式计算等场景。

RPC框架设计

由上面的RPC介绍可以看出，RPC的核心组件有

客户端(Client)：服务调用方。
客户端存根(Client Stub)：存放服务端地址信息，将客户端的请求参数数据信息打包成网络消息，再通过网络传输发送给服务端。
服务端存根(Server Stub)：接收客户端发送过来的请求消息并进行解包，然后再调用本地服务进行处理。
服务端(Server)：服务的真正提供者。
Network Service：底层传输，可以是 TCP 或 HTTP。

由这些组件相互协作完成跨进程的服务调用，基本流程如下

客户端（client）调用以本地调用方式调用服务；
客户端stub接收到调用后负责将方法、参数等组装成能够进行网络传输的消息体；
客户端stub找到服务地址，并将消息发送到服务端；
服务端stub收到消息后进行解码；
服务端stub根据解码结果调用本地的服务；
本地服务执行并将结果返回给服务端stub；
服务端stub将返回结果打包成消息并发送至消费方；
客户端stub接收到消息，并进行解码；
客户端Stub得到服务端返回的解决。
客户端收到最终结果。

RPC框架的目标就是要把2~8这些步骤都封装起来，让用户对这些细节透明。
要实现上面的整个流程，基于java语言，我们解决的技术问题有：

1. 序列化

序列化是指在上面的步骤中，客户端Stub，服务端Stub在收到上层返回的数据之后，需要通过我网络模块传输到对方进程时，数据出于性能，及安全的考虑，需要转化成方便网络传输的数据格式。从应用数据到网络传输数据的转化为序列化，网络传输数据到应用数据的转化成为反序列化。序列化在java里有默认实现，但是在目前大部分应用高并发的性能要求下，我一般采用其他高性能的序列化实现方案，比如，Protobuf、Kryo、Hessian、Jackson 等

2. 网络传输

网络传输我们都是基于OSI网络层的TCP/UDP socket实现，但是IO处理基于目前高并发的性能需求，传统的阻塞式IO已经不适用，java也提供了NIO，以及Java7的NIO.2实现

3. 注册中心

注册中心的作用，就是一个RPC服务的一个管理中心，它有两个主要的职能，第一个就是服务注册，对于服务端对外提供的RPC服务，要想被其他调用知道，并调用，需要把它的调用方法，统一发送给注册中心注册，包括寻址ID，入参回参类型等；第二个就是注册中心需要把注册到注册中心的服务端方法告知客户端，这样客户端才能实现调用。目前应用比较广泛的框架有 zookeeper，zookeeper维护一张服务注册表，让多个服务提供者形成一个集群。客户端通过注册表获得服务端访问地址（IP+端口），实现服务调用

4. 代理模式

RPC框架一个主要的特征就是封装了通信细节，使RPC接口调用像调用本地方法一样，要实现这一目标，就要用到代理模式，由rpc接口代理类实现底层的通信逻辑。java的代理实现主要有两种方法，java5以后的动态代理，还有Cglib字节码代理。

整合上面的RPC核心流程以及基于java spring框架，我们可以方便的实现RPC服务开发，其详细实现流程如下:

3.Dubbo RPC框架实现分析

Dubbo 框架简介

Dubbo是阿里开源的一款用于服务治理的RPC通信框架。Dubbo除了基础的RPC功能之外，还提供了服务治理能力，像，服务发现，服务降级，服务路由，集群容错。我们这里主要分析下Dubbo的RPC功能实现。

下面是Dubbo的架构：

这是Dubbo中各个角色的交互图，我们可以看到在Dubbo中不同的组件有不同的角色：服务提供者Provider、服务消费者Consumer、注册中心Registry、服务运行容器Container以及监控中心Monitor。这些角色在服务启动过程中会按照如下的过程交互：

容器Container启动服务提供者。
服务提供者Provider启动并向注册中心Registry注册服务。
服务消费者Consumer启动并从注册中心Registry订阅服务。
服务消费者Consumer向服务提供者Provider发起请求。
服务提供者和消费者同步监控中心统计信息。

Dubbo RPC模块的实现

dubbo-rpc模块中包含了多个包，其中dubbo-rpc-api包定义了一个具体的RPC实现需要实现的接口。通过明确定义RPC实现的接口，为自定义扩展提供了统一的API层。Dubbo的实现高度可扩展。

Dubbo支持协议扩展，这些扩展被放在dubbo-rpc模块的各个包中，比如：dubbp-rpc-dubbo包中包含了 dubbo协议 的扩展，dubbo-rpc-http包是HTTP协议实现的RPC扩展。dubbo-rpc模块中总共包含了13种不同协议实现的RPC扩展

我们可以基于技术栈的要求选用不同的RPC实现。官方建议如果是短报文的请求，dubbo协议是比较推荐的选择。下面，我们就以dubbo协议的RPC实现dubbo-rpc-dubbo为例，来看下Dubbo是如何对远程过程调用进行抽象的。

上图是Dubbo RPC 一个完整调用简要流程图，其中主要环节有调用的入口，从一个代理对象开始，然后走负载均衡策略选出调用的Invoker对象，再经过Filter链之后封装请求数据，之后走网络层，服务端获取网络请求中的Invoker对象，再走服务端Filter链，接着调用实际的接口实现代理，处理业务逻辑，然后按原路返回调用结果。

4.Dubbo RPC 关键源码分析

上图是dubbo rpc 完成一次RPC 方法调用所流经的主要组件

首先服务代理对象Proxy持有了一个Invoker对象。然后触发invoke调用。

java复制代码public class InvokerInvocationHandler implements InvocationHandler {
    private final Invoker<?> invoker;

    public InvokerInvocationHandler(Invoker<?> handler) {
        this.invoker = handler;
    }

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        ~~~
        /**
        * 此处为rpc服务调用的开始
        * 此时的invoker 为MockClusterInoker 
        * 在服务引用的时候包装
        */
        this.invoker.invoke(new RpcInvocation(method, args)).recreate()
        ~~~
    }
}

在invoke调用过程中，需要使用Cluster，Cluster负责容错。Cluster在调用之前会通过Directory获取所有可以调用的远程服务Invoker列表。

java复制代码public class MockClusterInvoker<T> implements Invoker<T> {
  ···
    public Result invoke(Invocation invocation) throws RpcException {
        Result result = null;
        //  从URL里面获取mock配置
        String value = this.directory.getUrl()
          .getMethodParameter(invocation.getMethodName(), "mock",Boolean.FALSE.toString()).trim();
        if (value.length() != 0 && !value.equalsIgnoreCase("false")) {
            if (value.startsWith("force")) {
              // 强制mock场景调用
                result = this.doMockInvoke(invocation, (RpcException)null);
            } else {
              //  失败进入mock场景
                try {
                    result = this.invoker.invoke(invocation);
                } catch (RpcException var5) {
                  //发生异常进入mock
                    result = this.doMockInvoke(invocation, var5);
                }
            }
        } else {
            //无mock场景直接调用
            result = this.invoker.invoke(invocation);
        }
        return result;
    }

获取到的 invoker列表和通过Dubbo SPI获取到的负载均衡器，Dubbo会调用 FailoverClusterInvoker的 doInvoke 方法。

java复制代码public abstract class AbstractClusterInvoker<T> implements Invoker<T> {
  ...
   public Result invoke(Invocation invocation) throws RpcException {
      // 检验节点是否销毁
        this.checkWhetherDestroyed();
        LoadBalance loadbalance = null;
      /**
      * 调用Directory -》 调用Rou
      * 生成invoker对象
      * 获取invoker列表
      ***/ 
        List<Invoker<T>> invokers = this.list(invocation);
        if (invokers != null && !invokers.isEmpty()) {
            // 获取第一个服务提供者的invoker的负载均衡策略，默认为random
            loadbalance = (LoadBalance)ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(((Invoker)invokers.get(0)).getUrl().getMethodParameter(RpcUtils.getMethodName(invocation), "loadbalance", "random"));
        }
        RpcUtils.attachInvocationIdIfAsync(this.getUrl(), invocation);
        return this.doInvoke(invocation, invokers, loadbalance);
    }
  ...
}

调用LoadBalance方法做负载均衡，最终选出一个可以调用的Invoker，调用invoker的invoke方法

java复制代码public class FailoverClusterInvoker<T> extends AbstractClusterInvoker<T> {
  public Result doInvoke(Invocation invocation, List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
        List<Invoker<T>> copyinvokers = invokers;
        // invoker空校验
        this.checkInvokers(invokers, invocation);
        // 获取重试次数，默认重试一次
        int len = this.getUrl().getMethodParameter(invocation.getMethodName(), "retries", 2) + 1;
        if (len <= 0) {
            len = 1;
        }
        RpcException le = null;
        List<Invoker<T>> invoked = new ArrayList(invokers.size());
        Set<String> providers = new HashSet(len);

        for(int i = 0; i < len; ++i) {
            if (i > 0) {
              // 有过一次失败之后，需要重新校验节点是否被销毁，invokkers是否为空
                this.checkWhetherDestroyed();
                copyinvokers = this.list(invocation);
                this.checkInvokers(copyinvokers, invocation);
            }
            // 重新负载均衡
            Invoker<T> invoker = this.select(loadbalance, invocation, copyinvokers, invoked);
            invoked.add(invoker);
            RpcContext.getContext().setInvokers(invoked);
            try {
              // 进行远程调用
                Result result = invoker.invoke(invocation);
                return var12;
            } catch (RpcException var17) {
                if (var17.isBiz()) {
                    throw var17;
                }
                le = var17;
            } catch (Throwable var18) {
                le = new RpcException(var18.getMessage(), var18);
            } finally {
                providers.add(invoker.getUrl().getAddress());
            }
        }
      ···
    }
}

上面的invoker对象实际上是InvokerWrapper类，内部包装了ProtocolFilterWrapper类,在这里会经过filter链和listener链的包装，在默认调用中会调用，ConsumerContextFilter，FutureFilter，MonitorFilter等过滤器，调用万filter链之后会调用listener链。

java复制代码private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
    Invoker<T> last = invoker;
    // 根据key和group获取自动激活的Filter
    List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
    if (!filters.isEmpty()) {
        // 这里是倒排遍历，因为只有倒排，最外层的Invoker才能使第一个过滤器
        for (int i = filters.size() - 1; i >= 0; i--) {
            final Filter filter = filters.get(i);
            // 会把真实的Invoker（服务对象ref）放到拦截器的末尾
            final Invoker<T> next = last;
            // 为每一个filter生成一个exporter，依次串起来
            last = new Invoker<T>() {
               ...
                @Override
                public Result invoke(Invocation invocation) throws RpcException {
                    // $-- 每次调用都会传递给下一个拦截器
                    return filter.invoke(next, invocation);
                }
                ...
            };
        }
    }
    return last;
}

java复制代码public class ConsumerContextFilter implements Filter {
    public ConsumerContextFilter() {
    }

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    RpcContext.getContext().setInvoker(invoker).setInvocation(invocation).setLocalAddress(NetUtils.getLocalHost(), 0).setRemoteAddress(invoker.getUrl().getHost(), invoker.getUrl().getPort());
        if (invocation instanceof RpcInvocation) {
            ((RpcInvocation)invocation).setInvoker(invoker);
        }

        RpcResult var4;
        try {
            RpcResult result = (RpcResult)invoker.invoke(invocation);
            RpcContext.getServerContext().setAttachments(result.getAttachments());
            var4 = result;
        } finally {
            RpcContext.getContext().clearAttachments();
        }

        return var4;
    }
}

包装类结束之后再调用AbstractInvoker的invoke方法，在这个方法里调用了DubboInvoker类的doInvoker方法

java复制代码public abstract class AbstractInvoker<T> implements Invoker<T> {
public Result invoke(Invocation inv) throws RpcException {
        if (this.destroyed.get()) {
           ···
        } else {
            RpcInvocation invocation = (RpcInvocation)inv;
            invocation.setInvoker(this);
            ···
            try {
                return this.doInvoke(invocation);
            } catch (InvocationTargetException var6) {
                ···
            } catch (RpcException var7) {
                ···
            } catch (Throwable var8) {
                return new RpcResult(var8);
            }
        }
    }
}

Dubbo协议相关部分的处理逻辑，包括请求的构建、传输。其下层就是Exchange层的内容

java复制代码public class DubboInvoker<T> extends AbstractInvoker<T> {
 protected Result doInvoke(Invocation invocation) throws Throwable {
        RpcInvocation inv = (RpcInvocation)invocation;
        String methodName = RpcUtils.getMethodName(invocation);
        inv.setAttachment("path", this.getUrl().getPath());
        inv.setAttachment("version", this.version);
        ExchangeClient currentClient;
        if (this.clients.length == 1) {
            currentClient = this.clients[0];
        } else {
            currentClient = this.clients[this.index.getAndIncrement() % this.clients.length];
        }

        try {
            boolean isAsync = RpcUtils.isAsync(this.getUrl(), invocation);
            boolean isOneway = RpcUtils.isOneway(this.getUrl(), invocation);
            int timeout = this.getUrl().getMethodParameter(methodName, "timeout", 1000);
            if (isOneway) {
                boolean isSent = this.getUrl().getMethodParameter(methodName, "sent", false);
                currentClient.send(inv, isSent);
                RpcContext.getContext().setFuture((Future)null);
                return new RpcResult();
            } else if (isAsync) {
                ResponseFuture future = currentClient.request(inv, timeout);
                RpcContext.getContext().setFuture(new FutureAdapter(future));
                return new RpcResult();
            } else {
                RpcContext.getContext().setFuture((Future)null);
                return (Result)currentClient.request(inv, timeout).get();
            }
        } catch (TimeoutException var9) {
          ···
        } catch (RemotingException var10) {
          ···
        }
    }
}

Dubbo的Exchange层和Transport层，主要是和网络传输、协议相关的内容

··· 忽略协议处理层内容

进入服务端ProtocolFilerWrapper类的invoker方法，开始服务端业务逻辑处理，这里首先还是要经已经包装好的filter链和listener链，最后会到真实方法的调用。

java复制代码@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
    try {
        Result result = invoker.invoke(invocation);
        // 不会处理GenericService类型invoker的异常
        if (result.hasException() && GenericService.class != invoker.getInterface()) {
        ····
        }
    }
}

经过InvokerWrapper包装类的方法。

11.在AbstractProxyInvoker.invoke 中会执行doInvoker方法。这里doInvoker的实现一般默认实现有两个JdkProxyFactory，JavassistProxyFactory，这里实际是获取实现类的代理，并执行代理类的业务方法，返回结果。

java复制代码/**
 * JdkRpcProxyFactory
 */
public class JdkProxyFactory extends AbstractProxyFactory {

    @Override
    @SuppressWarnings("unchecked")
    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        return (T) Proxy.newProxyInstance(invoker.getInterface().getClassLoader(), interfaces, new InvokerInvocationHandler(invoker));
    }

    @Override
    public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        return new AbstractProxyInvoker<T>(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class<?>[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                Method method = proxy.getClass().getMethod(methodName, parameterTypes);
                return method.invoke(proxy, arguments);
            }
        };
    }

}

总结下，整个Dubbo Rpc 服务调用的整个流程，在服务的引用过程中，Dubbo把Invoker层层包装为一个代理，用户使用这个代理对象像调用本地服务一样调用。Dubbo将代理的Invoker层层剥开，每层完成特定的功能，最终协调一致，完成了整个服务的调用。

本文转载自: 掘金

开发者博客 – 和开发相关的这里全都有