精华内容
下载资源
问答
  • Dubbo RPC

    2021-06-15 14:44:34
  • dubbo rpc

    2016-08-19 18:38:00
    本文基于dubbo-rpc-api dubbo-rpc-default两个包重点了解下dubbo rpc的实现原理。 #1、代码示例 既然要说原理,首先就是把实现了rpc的代码示例放在这里。 DemoService service = new DemoServiceImpl(); ...

    dubbo中rpc通信算是重头戏。本文基于dubbo-rpc-api dubbo-rpc-default两个包重点了解下dubbo rpc的实现原理。

    #1、代码示例 既然要说原理,首先就是把实现了rpc的代码示例放在这里。

    DemoService service = new DemoServiceImpl();
    		protocol.export(proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:9020/" + DemoService.class.getName() + "?codec=exchange")));
    		service = proxy.getProxy(protocol.refer(DemoService.class, URL.valueOf("dubbo://127.0.0.1:9020/" + DemoService.class.getName() + "?codec=exchange")));
    		assertEquals(service.getSize(new String[]{"", "", ""}), 3);
    

    将其“肢解”,我们就能够对rpc有了一定的了解。 #1、生成实例 DemoService service = new DemoServiceImpl(); 我们知道平时我们在spring里面配置dubbo信息到spring,客户端在只有接口的情况下就能够调到实现类,就好像这行代码一样。你有没有问,客户端本来没这个类怎么就能调到实现类的呢?代理呗,您如是回答。那么我们继续。 #2、服务端发布自己的服务 ##2.1、生成代理

    proxy.getInvoker(service, DemoService.class, URL.valueOf("dubbo://127.0.0.1:9020/" + DemoService.class.getName() + "?codec=exchange"))
    

    生成代理的过程就这么简单,而其中包含的东西真的不少,大致步骤: ###1、URL类会解析传递进来的字符串,根据规则将其放置到URL的各个字段里面去。 "dubbo://127.0.0.1:9020/" + DemoService.class.getName() + "?codec=exchange"是传递进来的字符串,仔细看它就是"dubbo://127.0.0.1:9020/DemoServie?codec=exchange". ###2、proxy.getInvoker(实例,接口,URL)做的事情比较多。 首先这个proxy是一个适配器,这个如果您跟代码会发现完全知道到,是因为这个适配器是在ExtendLoader里面动态生成的,在调用getInvoker方法的时候其实是在调用适配器里面的该方法。适配器里面的该方法会根据您传递进来的URL去寻找真正的实现类。这里找到类也许您会认为是JavassistProxyFactory,非也,其实找到的是StubProxyFactoryWrapper类,并把JavassistProxyFactory放置到该类中来。知道了这个类那么就该调用真正实现类的getInvoker方法。

      public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) throws RpcException {
            return proxyFactory.getInvoker(proxy, type, url);
        }
    

    这里就直接调用了JavassistProxyFactory里面的getInvoker方法。

      public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
            // TODO Wrapper类不能正确处理带$的类名
            final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
            return new AbstractProxyInvoker<T>(proxy, type, url) {
                @Override
                protected Object doInvoke(T proxy, String methodName, 
                                          Class<?>[] parameterTypes, 
                                          Object[] arguments) throws Throwable {
                    return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
                }
            };
        }
    

    但看这个方法貌似就做了两件事情:生成wrapper实例,返回一个AbstractProxyInvoker实例。这里面的wrapper真的是不能小觑,打开实现类会发现作者在这个类的代码量很大。那么他在做什么?

     ret = makeWrapper(c);
    

    它在包装实例。将之前的DemoServiceImpl实例包装成了一个wrapper返回。这个过程做了三件事情 首先:拼接了三个方法串。

    方法串1
     public void setPropertyValue(Object o, String n, Object v) {
            com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl w;
            try {
                w = ((com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl) $1);
            } catch (Throwable e) {
                throw new IllegalArgumentException(e);
            }
            throw new com.alibaba.dubbo.common.bytecode.NoSuchPropertyException("Not found property \"" + $2 + "\" filed or setter method in class com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl.");
        }
    
    

    方法串2
    public Object getPropertyValue(Object o, String n) {
            com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl w;
            try {
                w = ((com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl) $1);
            } catch (Throwable e) {
                throw new IllegalArgumentException(e);
            }
            if ($2.equals("threadName")) {
                return ($w) w.getThreadName();
            }
            throw new com.alibaba.dubbo.common.bytecode.NoSuchPropertyException("Not found property \"" + $2 + "\" filed or setter method in class com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl.");
        }
    

    方法串3
    
      public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException {
            com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl w;
            try {
                w = ((com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl) $1);
            } catch (Throwable e) {
                throw new IllegalArgumentException(e);
            }
            try {
                if ("invoke".equals($2) && $3.length == 2) {
                    return ($w) w.invoke((java.lang.String) $4[0], (java.lang.String) $4[1]);
                }
                if ("get".equals($2) && $3.length == 1) {
                    return ($w) w.get((com.alibaba.dubbo.rpc.protocol.dubbo.support.CustomArgument) $4[0]);
                }
                if ("timestamp".equals($2) && $3.length == 0) {
                    return ($w) w.timestamp();
                }
                if ("keys".equals($2) && $3.length == 1) {
                    return ($w) w.keys((java.util.Map) $4[0]);
                }
                if ("getSize".equals($2) && $3.length == 1 && $3[0].getName().equals("[Ljava.lang.String;")) {
                    return ($w) w.getSize((java.lang.String[]) $4[0]);
                }
                if ("getSize".equals($2) && $3.length == 1 && $3[0].getName().equals("[Ljava.lang.Object;")) {
                    return ($w) w.getSize((java.lang.Object[]) $4[0]);
                }
                if ("echo".equals($2) && $3.length == 1 && $3[0].getName().equals("java.util.Map")) {
                    return ($w) w.echo((java.util.Map) $4[0]);
                }
                if ("echo".equals($2) && $3.length == 1 && $3[0].getName().equals("java.lang.String")) {
                    return ($w) w.echo((java.lang.String) $4[0]);
                }
                if ("enumlength".equals($2) && $3.length == 1 && $3[0].getName().equals("[Lcom.alibaba.dubbo.rpc.protocol.dubbo.support.Type;")) {
                    return ($w) w.enumlength((com.alibaba.dubbo.rpc.protocol.dubbo.support.Type[]) $4[0]);
                }
                if ("enumlength".equals($2) && $3.length == 1 && $3[0].getName().equals("com.alibaba.dubbo.rpc.protocol.dubbo.support.Type")) {
                    return ($w) w.enumlength((com.alibaba.dubbo.rpc.protocol.dubbo.support.Type) $4[0]);
                }
                if ("stringLength".equals($2) && $3.length == 1) {
                    return ($w) w.stringLength((java.lang.String) $4[0]);
                }
                if ("sayHello".equals($2) && $3.length == 1) {
                    w.sayHello((java.lang.String) $4[0]);
                    return null;
                }
                if ("nonSerializedParameter".equals($2) && $3.length == 1) {
                    w.nonSerializedParameter((com.alibaba.dubbo.rpc.protocol.dubbo.support.NonSerialized) $4[0]);
                    return null;
                }
                if ("returnNonSerialized".equals($2) && $3.length == 0) {
                    return ($w) w.returnNonSerialized();
                }
                if ("getThreadName".equals($2) && $3.length == 0) {
                    return ($w) w.getThreadName();
                }
                if ("getbyte".equals($2) && $3.length == 1) {
                    return ($w) w.getbyte(((Byte) $4[0]).byteValue());
                }
                if ("gerPerson".equals($2) && $3.length == 1) {
                    return ($w) w.gerPerson((com.alibaba.dubbo.rpc.protocol.dubbo.support.Person) $4[0]);
                }
            } catch (Throwable e) {
                throw new java.lang.reflect.InvocationTargetException(e);
            }
            throw new com.alibaba.dubbo.common.bytecode.NoSuchMethodException("Not found method \"" + $2 + "\" in class com.alibaba.dubbo.rpc.protocol.dubbo.support.DemoServiceImpl.");
        }
    

    然后、生成实例

    ClassGenerator cc = ClassGenerator.newInstance(cl); 在这个实例里面加属性和方法(包括上面拼接的三个方法串)。 最后,生成cc的实例。将其转为Wrapper类型返回。 至此,我们可以这样理解,这样动态生成的Wrapper已经不是普通的Wrapper了,虽然普通的Wrapper也有方法invokeMethod。但是它并没有关联实际的实现类,作者在这块做了一个动态的Wrapper,代理了普通的Wrapper。 继续,别蒙,至此我们得到了一个AbstractProxyInvoker实例。 至此proxy.getInvoker()交代完毕。 ##2.2、protocol.export(Invoker<T> invoker) throws RpcException; 上文我们说这个参数invoker实际的类型是AbstractProxyInvoker。 接下来我们看看这个方法做什么了 ####1、protocol本身是protocol的适配器,会根据传入进来的URL找到真正实现类的Wrapper。这里为什么生成的是实现类的包装器,请读者去ExtendLoader里面一探究竟。Protocol的包装器类有以下两个 class com.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper class com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper 最终返回类型是ProtocolFilterWrapper

    
        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));
        }
        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.size() > 0) {
                for (int i = filters.size() - 1; i >= 0; i --) {
                    final Filter filter = filters.get(i);
                    final Invoker<T> next = last;
                    last = new Invoker<T>() {
    
                        public Class<T> getInterface() {
                            return invoker.getInterface();
                        }
    
                        public URL getUrl() {
                            return invoker.getUrl();
                        }
    
                        public boolean isAvailable() {
                            return invoker.isAvailable();
                        }
    
                        public Result invoke(Invocation invocation) throws RpcException {
                            return filter.invoke(next, invocation);
                        }
    
                        public void destroy() {
                            invoker.destroy();
                        }
    
                        @Override
                        public String toString() {
                            return invoker.toString();
                        }
                    };
                }
            }
            return last;
        }
    

    首先是创建一个过滤器链,将AbstractProxyInvoker实例层层包装 然后是调用protocol.export方法。 这个方法有是一个适配器代理找包装器的过程,最后发现没有包装器,直接返回了协议类DubboProtocol。该类的作用就是将Invoker转成Exporter,并调用dubbo根据这个url连接注册中心(或者直连)。 服务器端的内容就讲到这里。接下来我们关注客户端。 #3、客户端去获取服务 ##3.1、将请求的url包装成invoker protocol.refer(Class<T> type, URL url) 该方法返回ListenerInvokerWrapper类型的实例 ##3.2、获得invoker的代理。 proxy.getProxy(Invoker<T> invoker) throws RpcException proxy会根据ExtendLoader类获得这个类的wrapper:StubProxyFactoryWrapper。最终会调用该类的getProxy方法去获得代理。 1、调用具体的代理工厂创建一个invoker的代理类。这里面是JavassistProxyFactory类生成的代理。

      public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
            return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
        }
    

    ##3.3、调用接口方法 service.getSize(null)。 调用接口方法就是去执行上面代理类的过程,具体过程大体是: ###1、进入InvokerInvocationHandler的invoke方法 ###2、invoker方法发现我们调用的不是简单的方法(hashCode,equals)等,那么进入invoker。

    invoker.invoke(new RpcInvocation(method, args)).recreate()
    

    ####2.1、包装RpcInvocation类的实例。 ####2.2、调用ProtocolFilterWrapper类的invoke方法。 ####2.3、沿着过滤器链执行,知道执行真正的实现类invoker ####2.4、真正的invoker是DubboInvoker,它的doInvoke方法dubbo去找真正实现类去执行。这里面dubbo支持了同步和异步。

    至此,调用端的逻辑交代完毕。

    dubbo rpc的东西当然不止这些,其他别的东西还会陆续介绍。本文关注的是rpc 的主线,对代码的执行逻辑做一个交代。

    转载于:https://my.oschina.net/zjItLife/blog/735302

    展开全文
  • dubbo rpc Apache Software Foundation已将其基于Java的远程过程调用(RPC)框架Apache Dubbo从孵化器升级为顶级项目状态。 Dubbo包括基于接口的远程呼叫,容错,智能负载平衡以及自动服务注册和发现。 [ 15个Java...

    dubbo rpc

    Apache Software Foundation已将其基于Java的远程过程调用(RPC)框架Apache Dubbo从孵化器升级为顶级项目状态。 Dubbo包括基于接口的远程呼叫,容错,智能负载平衡以及自动服务注册和发现。

    [ 15个Java框架使开发人员受益匪浅 哪些工具支持Java的新模块化功能 | 通过InfoWorld的App Dev Report新闻通讯了解编程方面的热门话题。 ]

    像其他RPC系统一样,Dubbo基于定义服务的概念,即使用其参数和返回类型指定要远程调用的方法。 在服务器上,Dubbo实现该接口并侦听客户端调用。 在客户端上,存根提供与服务器相同的方法。 Dubbo的其他特定功能包括:

    • 支持多种负载平衡策略,下游服务状态用于减少延迟并提高系统吞吐量。
    • 支持可以离线或在线检测服务的服务注册表。
    • 微内核和插件设计可以由第三方实现扩展为跨核心功能,包括协议,序列化和传输。
    • 运行时流量路由,可以对其进行配置,以便可以根据规则路由流量。 这支持包括数据中心路由和蓝绿色部署的功能。 蓝绿色是针对开发人员的变更管理策略,类似于AB测试。
    • 可视化的服务治理,带有用于执行任务的工具,包括查询运行状况,服务元数据和统计信息。

    尽管Dubbo也可以在API配置中运行,但是运行Dubbo的最常见方式是在Spring Framework中 Dubbo由电子商务公司阿里巴巴(Alibaba)开源,并已被中国人寿(China Life)和中国电信(China Telecom)等其他公司使用。

    您可以从其Apache网站下载Apache Dubbo

    翻译自: https://www.infoworld.com/article/3397085/dubbo-java-rpc-project-graduates-to-apache-top-level-status.html

    dubbo rpc

    展开全文
  • Dubbo RPC框架原理解析和源码
  • Dubbo RPC原理解读

    万次阅读 2017-08-13 14:31:41
    探究dubbo rpc实现原理。 探究rpc从发出请求到收到返回结果这整个过程的详细过程。 学习rpc的负载均衡原理。 学习服务暴露、服务发现的原理以及实现细节。 多线程中dubbo是如何做到将返回结果和每个线程一一对应的。

    本文代码摘录的时候,将一些与本流程无关的内容去掉了,如有需要请看源码。

    一、   闲言碎语

    使用rpc框架已经多年了,虽然之前有研究过rpc的过程,但是却不曾详细阅读过dubbo的源码,探究过其中的设计思路与亮点。所以抽时间阅读了一下dubbo的源码,分享出来和大家一起学习。

     

    二、   目标与示例

    1.  目标

    l   探究dubbo rpc实现原理。

    l   探究rpc从发出请求到收到返回结果这整个过程的详细过程。

    l   学习rpc的负载均衡原理。

    l   学习服务暴露、服务发现的原理以及实现细节。

    l   多线程中dubbo是如何做到将返回结果和每个线程一一对应的。

     

    本文重点为源码分析和模型实现分析,如果对dubbo的概念和使用不熟悉,情移步官网。

    本文的所有分析均基于dubbo 2.5.3版本。

    本文假定使用zookeeper管理服务。

     

    2.  示例代码

    以下的分析基于以下配置方式。不同的配置方式并不会影响本文所需要解决的几个问题,只是一下方式配置会比较便于理解,所以这里依次做为示例。

    1)   consumer

    <bean id="rpcServiceRef" class="com.alibaba.dubbo.config.spring.ReferenceBean">
            <property name="interface" value="com.wzf.service.RpcService"/>
            <property name="application" ref="dubboApplicationConfig"/>
            <property name="registry" ref="dubboRegistryConfig"/>
            <property name="version" value="dev"/>
            <property name="timeout" value="3000"/>
            <property name="retries" value="0"/>
            <property name="check" value="false"/>
        </bean>


    2)   provider

        <bean id="rpcServiceExport" class="com.alibaba.dubbo.config.spring.ServiceBean">
            <property name="interface" value="com.wzf.funny.service.RpcService"/>
            <property name="ref" ref="rpcServiceImpl"/>
            <property name="application" ref="dubboApplicationConfig"/>
            <property name="registry" ref="dubboRegistryConfig"/>
            <property name="protocol" ref="dubboProtocolConfig"/>
            <property name="version" value="dev"/>
            <property name="timeout" value="0"/>
            <property name="retries" value="0"/>
        </bean>


    三、   服务调用模型

    1.  dubbo的模块模型

    dubbo的模块模型有些复杂,不太容易看懂,如果你也有同感的话,可以看一下本文后面的几部分,他们详细讲述了dubbo中rpc的调用链,其中包括了核心的几个类,比较便于理解。

     

    2.  服务调用关系模型


    如图所示,dubbo的RPC调用模型分为registry、provider、consumer、monitor这几个部分。此图展示了从服务注册、发现、调用的全过程,但dubbo是如何做到的呢?其实这个问题包括了以下几个问题:provider如何注册服务到zookeeper;consumer如何从zookeeper拉取provider信息;provider变化以后,zookeeper如何告知consumer;consumer如何调用provider。另外,监控逻辑很简单本文暂时不做分析。

     

    3.  Provider

    从源码上看ServiceBean主要完成以下几件工作:服务暴露,取消暴露服务。

    1)   暴露服务

    服务暴露开始于ServiceBean的afterPropertiesSet方法,此方法在ServiceBean的所有属性都被赋值以后被BeanFactory调用。服务暴露的调用链是: ServiceConfig#export-> ServiceConfig#doExport -> ServiceConfig#doExportUrls -> ServiceConfig#doExportUrlsFor1Protocol-> ServiceConfig#exportLocal(URL url)。 暴露服务其实包括两个类容:

    l   将Invoker存入AbstractProtocol#exporterMap,调用服务时从次map中取出Invoker直接使用。

    protected finalSet<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();

    其中key为:com.wzf.funny.service.ArticleService:dev,value为invoker对象

    l   将url注册到zookeeper。

    此过程的入口在RegistryProtocol#export方法中,调用链为:

    RegistryProtocol#export -> FailbackRegistry#register ->AbstractRegistry#register -> ZookeeperRegistry#doRegister -> ZookeeperClient#create-> AbstractZookeeperClient#create

     

    2)   服务发现

    ZookeeperRegistry是服务发现的核心类之一,实现了《服务调用关系模型》中的register、subscribe、notify。以下分析一下几个主要的方法。

    l  构造函数

    从以下代码中可以看到,zkClient创建成功以后,会监听RECONNECTED事件,recover方法主要做一件事:将需要暴露的url放在failedRegistered(Set<URL>)中,将需要订阅的服务放在failedSubscribed(Set<URL>)中。说明RECONNECTED时,因为所有需要暴露的服务都需要重新注册,所以其实是将需要暴露、订阅的url都放到failedRegistered、failedSubscribed中。

    public ZookeeperRegistry(URL url,ZookeeperTransporter zookeeperTransporter) {
           super(url);
           //其他代码省略
           this.root = group;
            zkClient = zookeeperTransporter.connect(url);
           zkClient.addStateListener(new StateListener() {
               public void stateChanged(intstate) {
                    if (state == RECONNECTED) {
                          try {
                                    recover();
                              } catch (Exception e){
                                    logger.error(e.getMessage(),e);
                              }
                    }
               }
           });
    }

    创建zkclient的url示例如下:

    zookeeper://localhost:2181/com.alibaba.dubbo.registry.RegistryService?application=funny&dubbo=2.5.3&interface=com.alibaba.dubbo.registry.RegistryService&pid=38796&timestamp=1502594657663

     

    l  register(URL url)

    注册url代表的服务到zookeeper

    l  unregister(URLurl) 

    从zookeeper中删除之前注册的服务

    l  subscribe(URL url,NotifyListener listener)

    订阅url的服务

    l  unsubscribe(URL url,NotifyListener listener)

    取消订阅url对应的服务

    l  notify(URL url,NotifyListener listener, List<URL> urls)

    通知

    l  retry()

    上面提到过,在recover()中将需要暴露的服务放到failedRegistered(Set<URL>)中,将需要订阅的服务放在failedSubscribed(Set<URL>)中,并没有真正的重新暴露服务或者订阅服务,这个工作是放在retry()中的,另外notify、doUnsubscribe,failedUnregistered也都放在此方法中处理。retry()方法的主要逻辑如下(为了方便阅读,我删掉了部分代码),retry被一个定时线程调用:

       protected voidretry() {
           if (! failedRegistered.isEmpty()) {
               for (URL url : failed) {
                  doRegister(url);
                  failedRegistered.remove(url);
               }
           }
           if(! failedUnregistered.isEmpty()) {
               for (URL url : failed) {
                  doUnregister(url);
                  failedUnregistered.remove(url);
               }
           }
           if (! failedSubscribed.isEmpty()) {
              for(Map.Entry<URL, Set<NotifyListener>> entry: failed.entrySet()) {
                  URL url= entry.getKey();
                  Set<NotifyListener> listeners = entry.getValue();
                  for(NotifyListener listener : listeners) {
                      doSubscribe(url, listener);
                      listeners.remove(listener);
                  }
               }
           }
           if (! failedUnsubscribed.isEmpty()) {
              for(Map.Entry<URL, Set<NotifyListener>> entry: failed.entrySet()) {
                  URL url= entry.getKey();
                  Set<NotifyListener> listeners = entry.getValue();
                  for(NotifyListener listener : listeners) {
                      doUnsubscribe(url, listener);
                      listeners.remove(listener);
                  }
               }
           }
           if (! failedNotified.isEmpty()) {
              for(Map<NotifyListener, List<URL>> values: failed.values()) {
                 for(Map.Entry<NotifyListener, List<URL>> entry:values.entrySet()) {
                    NotifyListener listener = entry.getKey();
                    List<URL> urls = entry.getValue();
                    listener.notify(urls);
                    values.remove(listener);
                 }
               }
           }
       }
     
         this.retryFuture= retryExecutor.scheduleWithFixedDelay(new Runnable() {
               public void run() {
                   //检测并连接注册中心
                   try{
                       retry();
                   } catch(Throwable t) { //防御性容错
                       logger.error("Unexpected error occur at failed retry,cause: " + t.getMessage(), t);
                   }
               }
           }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS); 

    3)   取消暴露服务

    取消服务暴露是将服务从zk中移除的过程,保证此后consumer无法再使用此服务。ZkclientZookeeperClient中订阅了所有状态改变的事件,状态的改变最终会触发调用recover方法,从而导致调用unRegister方法,将zk节点删除。

    另外,因为在zk那边建立的是临时的节点,所以服务器和zk断开联系后,node将自动删除。Consumer将收到notify消息。

         publicZkclientZookeeperClient(URL url) {
               super(url);
               client= new ZkClient(url.getBackupAddress());
               client.subscribeStateChanges(new IZkStateListener() {
                    publicvoid handleStateChanged(KeeperState state) throwsException {
                         ZkclientZookeeperClient.this.state = state;
                         if(state == KeeperState.Disconnected) {
                              stateChanged(StateListener.DISCONNECTED);
                         } elseif (state== KeeperState.SyncConnected) {
                              stateChanged(StateListener.CONNECTED);
                         }
                    }
                    publicvoid handleNewSession() throws Exception {
                         stateChanged(StateListener.RECONNECTED);
                    }
               });
         }

    4)   RPC调用

    l   Wrapper类

    调用Wrapper#getWrapper方法时,会尝试从WRAPPER_MAP中获取,如果获取到直接返回,如果获取不到,则进入makeWrapper方法创建一个,创建好了以后放入WRAPPER_MAP中。makeWrapper是一个核心的方法,这个方法中做对原有RpcService的封装,具体逻辑如下。

    首先创建三个方法:setPropertyValue、getPropertyValue、invokeMethod,代码如下

         StringBuilder c1 = newStringBuilder("public void setPropertyValue(Object o, String n, Objectv){ ");
         StringBuilder c2 = newStringBuilder("public Object getPropertyValue(Object o, String n){ ");
         StringBuilder c3 = newStringBuilder("public Object invokeMethod(Object o, String n, Class[] p,Object[] v) throws " + InvocationTargetException.class.getName() + "{"};

    然后遍历RpcService的所有属性、方法,在原有属性的get、set、invoke的时候添加一些逻辑,因为invokeMethod方法与rpc关系最为密切的方法,所以重点讨论此方法。生成invokemethod的逻辑就是一个字符串拼接的过程,就不讨论了,这里将结果贴出来讨论一下,如下。其中:$1表示proxy$2表示methodName;$3表示parameterTypes;$4表示arguments;$w表示returnType

         publicObject invokeMethod(Object o, String n, Class[] p,Object[] v)
                    throwsjava.lang.reflect.InvocationTargetException {
               com.wzf.funny.service.ArticleService w;
               try{
                    w= ((com.wzf.funny.service.ArticleService) $1);
               } catch(Throwable e) {
                    thrownew IllegalArgumentException(e);
               }
               try{
                    if("pageQuery".equals($2)&& $3.length == 2) {
                         return($w) w.pageQuery((com.wzf.funny.query.ArticleQuery)$4[0], ((Boolean) $4[1]).booleanValue());
                    }
               } catch(Throwable e) {
                    thrownewjava.lang.reflect.InvocationTargetException(e);
               }
               thrownewcom.alibaba.dubbo.common.bytecode.NoSuchMethodException(
                         "Notfound method \"" + $2 + "\" in classcom.wzf.funny.service.ArticleService.");
         }


     

    最后构建Wrapper对象,构建的时候加上一些属性、方法,其中c1表示setPropertyValue,c2表示getPropertyValue ,c3表示invokeMethod。代码如下:

     

        ClassGenerator cc= ClassGenerator.newInstance(cl);
         cc.setClassName(( Modifier.isPublic(c.getModifiers()) ? Wrapper.class.getName() : c.getName()+ "$sw" ) + id );
         cc.setSuperClass(Wrapper.class);
         cc.addDefaultConstructor();
         cc.addField("publicstatic String[] pns;"); // property name array.
         cc.addField("public static " + Map.class.getName() + "pts;"); // property type map.
         cc.addField("publicstatic String[] mns;"); // all method name array.
         cc.addField("publicstatic String[] dmns;"); // declared method name array.
         for(int i=0,len=ms.size();i<len;i++)
              cc.addField("public static Class[] mts" + i + ";");
         cc.addMethod("publicString[] getPropertyNames(){ return pns; }");
         cc.addMethod("publicboolean hasProperty(String n){ return pts.containsKey($1); }");
         cc.addMethod("publicClass getPropertyType(String n){ return (Class)pts.get($1); }");
         cc.addMethod("publicString[] getMethodNames(){ return mns; }");
         cc.addMethod("publicString[] getDeclaredMethodNames(){ return dmns; }");
         cc.addMethod(c1.toString());
         cc.addMethod(c2.toString());
         cc.addMethod(c3.toString());
         Class<?> wc= cc.toClass();
         // setup static field.
         wc.getField("pts").set(null,pts);
         wc.getField("pns").set(null,pts.keySet().toArray(new String[0]));
         wc.getField("mns").set(null,mns.toArray(newString[0]));
         wc.getField("dmns").set(null,dmns.toArray(newString[0]));
         int ix = 0;
         for( Methodm : ms.values())
              wc.getField("mts"+ ix++).set(null,m.getParameterTypes());
         return(Wrapper)wc.newInstance();

     

    l   JavassistProxyFactory#getInvoker

    如下代码所示,在JavassistProxyFactory中创建Invoker时,其实创建的是AbstractProxyInvoker的子类,其中proxy为xml中配置的rpcServiceImpl对象,即我们的目标对象。当consumer发起Rpc请求时,会将classname、methodname、 parameterTypes、arguments这些数据传输过来,在wrapper.invokeMethod中通过动态代理技术,直接调用rpcServiceImpl中的 methodname方法。

       public <T> Invoker<T>getInvoker(T proxy, Class<T> type, URL url){
            // TODO Wrapper类不能正确处理带$的类名
           final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass(): type);
           return new AbstractProxyInvoker<T>(proxy, type, url) {
               @Override
               protectedObject doInvoke(T proxy, String methodName,   Class<?>[] parameterTypes, Object[] arguments) throwsThrowable {
                    return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
               }
           };
       }
    

     

    4.  Consumer

    消费的核心是负载均衡,和单nio通道在多线程情况下的实现逻辑。

    1)   负载均衡算法

    l   RandomLoadBalance

    先计算是否所有invoker的权重是否相同,相同则直接random一下,否则根据权重加权。主要代码如下:

           if (totalWeight > 0 && ! sameWeight) {
                // 如果权重不相同且权重大于0则按总权重数随机
               int offset = random.nextInt(totalWeight);
                // 并确定随机值落在哪个片断上
               for (int i = 0; i < length; i++) {
                   offset-= getWeight(invokers.get(i), invocation);
                   if(offset < 0) {
                       return invokers.get(i);
                   }
               }
           }
            // 如果权重相同或权重为0则均等随机
           return invokers.get(random.nextInt(length));

     

    l   LeastActiveLoadBalance

    先计算出一个最少活跃数的invoker集合,然后从这个集合中随机选取一个,然后计算是否所有invoker的权重是否相同,相同则直接random一下,否则根据权重加权取invoker。代码如下:

           int length = invokers.size(); // 总个数
           int leastActive = -1; // 最小的活跃数
           int leastCount = 0; // 相同最小活跃数的个数
           int[] leastIndexs = newint[length];// 相同最小活跃数的下标
           int totalWeight = 0; // 总权重
            int firstWeight = 0; // 第一个权重,用于于计算是否相同
           boolean sameWeight = true;// 是否所有权重相同
           for (int i = 0; i < length; i++) {
               Invoker<T> invoker = invokers.get(i);
               int active = RpcStatus.getStatus(invoker.getUrl(), invocation.getMethodName()).getActive();// 活跃数
               int weight = invoker.getUrl().getMethodParameter(invocation.getMethodName(), Constants.WEIGHT_KEY, Constants.DEFAULT_WEIGHT);// 权重
               if (leastActive == -1 || active< leastActive) { //发现更小的活跃数,重新开始
                   leastActive= active; // 记录最小活跃数
                   leastCount= 1; // 重新统计相同最小活跃数的个数
                   leastIndexs[0]= i; // 重新记录最小活跃数下标
                   totalWeight= weight; // 重新累计总权重
                   firstWeight= weight; // 记录第一个权重
                   sameWeight= true; // 还原权重相同标识
               } else if (active == leastActive) { // 累计相同最小的活跃数
                   leastIndexs[leastCount ++] = i;// 累计相同最小活跃数下标
                   totalWeight+= weight; // 累计总权重
                   //判断所有权重是否一样
                   if(sameWeight && i > 0 
                          && weight != firstWeight) {
                       sameWeight = false;
                   }
               }
           }
            // assert(leastCount> 0)
           if (leastCount == 1) {
                // 如果只有一个最小则直接返回
               return invokers.get(leastIndexs[0]);
           }
           if (! sameWeight && totalWeight> 0) {
                // 如果权重不相同且权重大于0则按总权重数随机
               int offsetWeight = random.nextInt(totalWeight);
                // 并确定随机值落在哪个片断上
               for (int i = 0; i < leastCount;i++) {
                   intleastIndex = leastIndexs[i];
                   offsetWeight-= getWeight(invokers.get(leastIndex), invocation);
                   if(offsetWeight <= 0)
                       return invokers.get(leastIndex);
               }
           }
            // 如果权重相同或权重为0则均等随机
            return invokers.get(leastIndexs[random.nextInt(leastCount)]);


     

    l   RoundRobinLoadBalance

    记录一个调用次数的数字,然后每次调用时对总invoker取模,并在调用次数基础上自增;权重不同的时候,逻辑稍有不同,具体可以参考远嘛。主要代码如下:

           AtomicPositiveInteger sequence= sequences.get(key);
           if (sequence == null){
               sequences.putIfAbsent(key, newAtomicPositiveInteger());
               sequence= sequences.get(key);
           }
           // 取模轮循
           return invokers.get(sequence.getAndIncrement()% length);

    l   ConsistentHashLoadBalance

    计算一致性hash的值,然后选取invoker。代码如下:

           String key = invokers.get(0).getUrl().getServiceKey() + "." + invocation.getMethodName();
           int identityHashCode = System.identityHashCode(invokers);
           ConsistentHashSelector<T> selector= (ConsistentHashSelector<T>) selectors.get(key);
           if (selector == null|| selector.getIdentityHashCode() != identityHashCode) {
               selectors.put(key, newConsistentHashSelector<T>(invokers, invocation.getMethodName(), identityHashCode));
               selector= (ConsistentHashSelector<T>) selectors.get(key);
           }
           return selector.select(invocation);



    2)   Invoker

    l FactoryBean创建proxy的调用链:

    ReferenceBean#getObject–>ReferenceBean#get–>ReferenceConfig#init–>ReferenceBean#createProxy。

    在ReferenceBean#createProxy()方法中创建Invoker;通过Invoker创建proxy。

    l 创建Invoker,并向zk注册监听的consumer。

    RegistryProtocol#doRefer–>RegistryProtocol#doRefer–>FailbackRegistry#register–>FailbackRegistry#doRegister–>ZookeeperRegistry#doRegister–>zkClient#create

    在RegistryProtocol#doRefer方法中,除了调用FailbackRegistry#register注册服务以外,还会调用RegistryDirectory#subscribe来订阅此服务,次操作会注册Listener。

    Consumer url示例:

    consumer://192.168.222.34/com.wzf.funny.service.RpcService?application=weixin&category=consumers&check=false&dubbo=2.5.3&interface=com.wzf.funny.service.RpcService&methods=sayHello&pid=44244&retries=0&revision=0.1.0-SNAPSHOT&side=consumer&timeout=5000&timestamp=1502795345908&version=dev

     

    3)   InvokerInvocationHandler

    示例代码中com.alibaba.dubbo.config.spring.ReferenceBean是一个FactoryBean,通过context.getBean方法获取的是ReferenceBean#getObject方法的返回结果,ReferenceBean#getObject()方法返回的是一个proxy对象,此proxy持有一个InvokerInvocationHandler属性,如下图所示

    rpc调用示例代码如下:

    rpcService.sayHello()

    rpcService是一个proxy对象(ReferenceBean#getObject()返回的对象),当调用sayHello()方法时,最终由InvokerInvocationHandler#invoker处理。


    5   多线程下的通信

    DubboInvoker#doInvoke方法中,在ExchangeClient#request(inv,timeout)调用时,返回一个DefaultFuture对象,接着会调用DefaultFuture.get()方法(等待返回结果)。

    对于consumer端而言,服务器会为每一个请求创建一个线程,因为rpc操作是一个慢动作,为了节省资源,当线程发送rpc请求后,需要让当前线程释放资源、进入等待队列,当获取到返回结果以后,再唤醒这个线程。

    RPC请求的过程为:每一个RPC请求都有一个唯一的id,RPC请求的时候,会将此id也发送给provider;provider处理完请求后会将此id和返回结果一同返回给consumer;consumer收到返回信息以后解析出id,然后从FUTURES中找到相对应的DefaultFuture,并通过DefaultFuture.done#signal()唤醒之前等待线程。

    下面根据源码详细讨论一下多线程情况下rpc请求的细节,即dubbo多线程模型的实现。

    l   DefaultFuture#field

    这里列出了与多线程相关的几个重要的属性

       private finalLock                           lock = new ReentrantLock();
       private finalCondition                      done = lock.newCondition();
       private staticfinal Map<Long, DefaultFuture> FUTURES   = newConcurrentHashMap<Long, DefaultFuture>();

     

    l   DefaultFuture#构造函数

    创建好DefaultFuture对象以后,将DefaultFuture存入了FUTURES中。其实每一次请求,多会生成一个唯一的id,即对于每个服务器而言,id唯一。

     public DefaultFuture(Channel channel, Request request,int timeout){
           this.channel = channel;
           this.request = request;
           this.id = request.getId();
           this.timeout = timeout> 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
            // put into waitingmap.
           FUTURES.put(id, this);
           CHANNELS.put(id, channel);
       }

     

    l   DefaultFuture#get

    主要逻辑是:获取锁,调用await方法,此时当前线程进入等待队列,此线程会有两种结果过:要么超时,要么被唤醒;如果被唤醒,则返回rpc的结果。

      public Object get(int timeout) throws RemotingException {
           if (timeout <= 0) {
               timeout =Constants.DEFAULT_TIMEOUT;
           }
           if (! isDone()) {
               long start = System.currentTimeMillis();
               lock.lock();
               try {
                   while(! isDone()) {
                       done.await(timeout,TimeUnit.MILLISECONDS);
                       if (isDone() || System.currentTimeMillis() - start > timeout){
                           break;
                       }
                   }
               } catch(InterruptedException e) {
                   thrownew RuntimeException(e);
               } finally{
                   lock.unlock();
               }
               if (!isDone()) {
                   thrownew TimeoutException(sent > 0, channel,getTimeoutMessage(false));
               }
           }
           returnreturnFromResponse();
       }


    l   DefaultFuture#received

    收到返回结果时,调用此方法。首先从FUTURES中根据id获取DefaultFuture,如果不存在,打印一条日志;如果存在则通过signal释放一个唤醒信号,将线程从等待队列中唤醒。

        public staticvoid received(Channel channel, Response response){
           try {
               DefaultFuture future= FUTURES.remove(response.getId());
               if (future != null){
                   future.doReceived(response);
               } else {
                    logger.warn("Thetimeout response finally returned at ")。
               }
           } finally {
               CHANNELS.remove(response.getId());
           }
       }
     
       private voiddoReceived(Response res) {
           lock.lock();
           try {
               response= res;
               if (done != null) {
                   done.signal();
               }
           } finally {
               lock.unlock();
           }
           if (callback != null){
               invokeCallback(callback);
           }
       }


    DefaultFuture#received

    以下代码是用来从FUTURES清理rpc请求超时的DefaultFuture

       private staticclass RemotingInvocationTimeoutScan implements Runnable {
           public void run() {
               while (true) {
                   try{
                       for (DefaultFuture future: FUTURES.values()) {
                           if (future == null || future.isDone()){
                              continue;
                           }
                           if (System.currentTimeMillis() - future.getStartTimestamp() > future.getTimeout()) {
                               // create exception response.
                              Response timeoutResponse = new Response(future.getId());
                              // set timeout status.
                              timeoutResponse.setStatus(future.isSent() ? Response.SERVER_TIMEOUT : Response.CLIENT_TIMEOUT);
                              timeoutResponse.setErrorMessage(future.getTimeoutMessage(true));
                              // handle response.
                              DefaultFuture.received(future.getChannel(),timeoutResponse);
                           }
                       }
                       Thread.sleep(30);
                   } catch(Throwable e) {
                       logger.error("Exception when scan the timeoutinvocation of remoting.", e);
                   }
               }
           }
       }
       static {
           Thread th = new Thread(newRemotingInvocationTimeoutScan(), "DubboResponseTimeoutScanTimer");
           th.setDaemon(true);
           th.start();
       }

    四、   RPC调用流程

    1.  简化流程图


    此流程图是一个简化的流程图,主要描述dubbo调用的全过程。

    RPC调用时,Consumer根据负载均衡算法获取invoker,在执行完filter链以后,就开始平装数据,发送数据到socket中,consumer这一端通过ReentrantLock进入await状态。

    Provider从socket获取数据后,执行receive方法, 接着执行Filter链,接着找到invoker通过代理对象执行Service,最后将返回结果写入socket。

    Consumer收到返回结果以后,唤醒之前await的内容,然后将返回结果返回给调用方。

     

    2.  完整流程图


    如上图所示,这是一个完整的调用流程图,包括了执行过程中主要的类和方法。

    后续内容主要是对次流程图的详细描述,如果次流程图已经完全清晰,可以忽略后面的内容。

     

    五、   consumer端远程调用请求链

    对远程方法的调用,其实是对InvokerInvocationHandler#invoke的调用。

    1.  InvokerHandler

    1)    InvokerInvocationHandler#invoke

    对执行方法做简单处理(toString、hashCode、equals这些方法不调用远程接口)后,执行MockClusterInvoker#invoke方法。

           if (method.getDeclaringClass() == Object.class) {
               return method.invoke(invoker,args);
           }
           if ("toString".equals(methodName) && parameterTypes.length == 0) {
               return invoker.toString();
           }
           if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
               return invoker.hashCode();
           }
           if ("equals".equals(methodName)&& parameterTypes.length == 1) {
               return invoker.equals(args[0]);
           }
            returninvoker.invoke(newRpcInvocation(method, args)).recreate();


    3)    MockClusterInvoker#invoke

    检查是否是mock,如果是,mock返回结果;如果不是的话进入FailoverClusterInvoker#invoke方法。

           String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim(); 
           if (value.length() == 0 || value.equalsIgnoreCase("false")){
           //no mock
           result = this.invoker.invoke(invocation);
           } else if (value.startsWith("force")) {
            //force:directmock
           result = doMockInvoke(invocation,null);
           } else {
            //fail-mock
                 result = this.invoker.invoke(invocation);
           }


    2.  解析出loadBalance,通过loadBalance算法获取Invoker对象。

    1)    FailoverClusterInvoker#invoke方法

     

    先执行父类AbstractClusterInvoker#invoke方法,获取List<Invoker<T>> invokers,loadBanlace;然后调用FailoverClusterInvoker#doInvoke方法。

           LoadBalance loadbalance;
           List<Invoker<T>> invokers= list(invocation);
           if (invokers != null&& invokers.size() > 0) {
               loadbalance= ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl().getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
           } else {
               loadbalance= ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
           }
           RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
            return doInvoke(invocation, invokers, loadbalance);

    2)    FailoverClusterInvoker#doInvoke

    循环1+retries次,知道成功或者重试次数耗尽,每次都先通过父类AbstractClusterInvoker#doselect方法获取invoker;然后执行invoker.invoke(),这个方法会进入一个调用链。

    其中需要注意的是,重试获取invoker的时候,需要间检查是否有invokers被销毁,invokers是否都可用

           checkWheatherDestoried();

           copyinvokers = list(invocation);

          //重新检查一下

          checkInvokers(copyinvokers,invocation);

     

    3)    AbstractClusterInvoker#doselect

    如果没有可用的invoker,直接返回;如果只有一个invoker,那么直接返回;如果有两个invoker,改成轮询算法,即如果上次使用了invokers.get(0),这次就直接使用invokers.get(1);如果有更多invoker,则通过loadBalance进行选择;如果之前的选中列表中已经包含了此次选中的invoker,那么重新选择。Dubbo默认使用random方式进行负载均衡。

            if (invokers == null|| invokers.size() == 0)
               return null;
           if (invokers.size() == 1)
               return invokers.get(0);
            // 如果只有两个invoker,退化成轮循
           if (invokers.size() == 2 && selected != null&& selected.size() > 0) {
               return selected.get(0) == invokers.get(0)? invokers.get(1) : invokers.get(0);
           }
           Invoker<T> invoker= loadbalance.select(invokers, getUrl(), invocation);
           
            //如果selected中包含(优先判断) 或者 不可用&&availablecheck=true 则重试.
           if( (selected != null&& selected.contains(invoker))  ||(!invoker.isAvailable()&& getUrl()!=null && availablecheck)){
               try{
                   Invoker<T> rinvoker = reselect(loadbalance,invocation, invokers,selected, availablecheck);
                   if(rinvoker != null){
                       invoker =  rinvoker;
                   }else{
                       //看下第一次选的位置,如果不是最后,选+1位置.
                       int index = invokers.indexOf(invoker);
                       try{
                           //最后在避免碰撞
                           invoker = index<invokers.size()-1?invokers.get(index+1):invoker;
                       }catch (Exception e){
                           logger.warn(e.getMessage()+" maybecause invokers list dynamic change, ignore.",e);
                       }
                   }
               }catch(Throwable t){
                    logger.error("clustorrelselect fail reason is :"+t.getMessage() +" if can notslove ,you can set cluster.availablecheck=false in url",t);
               }
           }
    
     

    4)    RandomLoadBalance#doSelect

    先根据定义的weight(默认为100)对每个invoker进行加权,然后随机取出一个。

           int length = invokers.size(); // 总个数
           int totalWeight = 0; // 总权重
           boolean sameWeight = true;// 权重是否都一样
           for (int i = 0; i < length; i++) {
               int weight = getWeight(invokers.get(i), invocation);
               totalWeight+= weight; // 累计总权重
               if (sameWeight && i> 0
                       && weight != getWeight(invokers.get(i - 1), invocation)){
                   sameWeight= false; // 计算所有权重是否一样
               }
           }
           if (totalWeight > 0 && ! sameWeight) {
                // 如果权重不相同且权重大于0则按总权重数随机
               int offset = random.nextInt(totalWeight);
                // 并确定随机值落在哪个片断上
               for (int i = 0; i < length; i++) {
                   offset-= getWeight(invokers.get(i), invocation);
                   if(offset < 0) {
                       return invokers.get(i);
                   }
               }
           }
            // 如果权重相同或权重为0则均等随机
           return invokers.get(random.nextInt(length));

    3.  执行invoke的Filter链。

    调用链是在xml加载的时候注册进来的;执行时按照以下顺序执行调用链中的invoke方法。

    com.alibaba.dubbo.rpc.filter.ConsumerContextFilter,
    com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter,
    com.alibaba.dubbo.monitor.support.MonitorFilter

    4.  执行invoke逻辑。

    1)    AbstractInvoker#invoke

    设置invocation信息,包括invoker、interface、sync、context等。

           RpcInvocation invocation= (RpcInvocation) inv;
           invocation.setInvoker(this);
           if (attachment != null&& attachment.size() > 0) {
           invocation.addAttachmentsIfAbsent(attachment);
           }
           Map<String, String> context= RpcContext.getContext().getAttachments();
           if (context != null){
           invocation.addAttachmentsIfAbsent(context);
           }
           if(getUrl().getMethodParameter(invocation.getMethodName(),Constants.ASYNC_KEY, false)){
           invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
           }
           RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
           return doInvoke(invocation);

    2)    DubboInvoker#doInvoke

    设置Invocation的属性,获取ExchangeClient,并执行request请求。

           RpcInvocation inv =(RpcInvocation) invocation;
           final String methodName = RpcUtils.getMethodName(invocation);
           inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
           inv.setAttachment(Constants.VERSION_KEY, version);
           
           ExchangeClient currentClient;
           if (clients.length== 1) {
               currentClient= clients[0];
           } else {
               currentClient= clients[index.getAndIncrement()% clients.length];
           }
           RpcContext.getContext().setFuture(null);
           return (Result) currentClient.request(inv,timeout).get();
     

    3)    DefaultFuture#get

    进入await,等待provider返回结果。

           if (timeout <= 0) {
               timeout =Constants.DEFAULT_TIMEOUT;
           }
           if (! isDone()) {
               long start = System.currentTimeMillis();
               lock.lock();
               try {
                   while(! isDone()) {
                       done.await(timeout,TimeUnit.MILLISECONDS);
                       if (isDone() || System.currentTimeMillis() - start > timeout){
                           break;
                       }
                   }
               } catch(InterruptedException e) {
                   thrownew RuntimeException(e);
               } finally{
                   lock.unlock();
               }
               if (!isDone()) {
                   thrownew TimeoutException(sent > 0, channel,getTimeoutMessage(false));
               }
           }
           returnreturnFromResponse(); 

    5.  request远程数据

    1)    HeaderExchangeChannel#requst

    创建Reuqest,并通过NettyClient发送请求。在创建DefaultFuture时,会将次DefaultFuture放入FUTURES(一个ConcurrentHashMap)中,也会将Channel放入CHANNELS(一个ConcurrentHashMap)中。

           Request req = new Request();
           req.setVersion("2.0.0");
           req.setTwoWay(true);
           req.setData(request);
           DefaultFuture future = new DefaultFuture(channel,req, timeout);
           channel.send(req);
           return future; 

    2)    NettyClient#request

    获取NettyChannel,并通过NettyChannel发送消息

           Channel channel =getChannel();
           if (channel == null|| ! channel.isConnected()) {
              throw newRemotingException(this, "message can not send, becausechannel is closed . url:" + getUrl());
           }
            channel.send(message, sent);

    3)    NettyChannel#send

    通过NioClientSocketChannel.write将数据通过socket发送出去。

           boolean success = true;
           int timeout = 0;
           try {
               ChannelFuture future= channel.write(message);
               if (sent) {
                   timeout= getUrl().getPositiveParameter(Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
                   success= future.await(timeout);
               }
               Throwable cause= future.getCause();
               if (cause != null){
                   throwcause;
               }
           } catch (Throwable e) {
               throw new RemotingException(this,"Failed to send message " + message + " to" + getRemoteAddress() + ", cause:" + e.getMessage(), e);
           }

     

    6.  常见消息格式

    Request [id=1, version=2.0.0,twoway=true, event=false, broken=false, data=RpcInvocation[methodName=generateArticle, parameterTypes=[long, class java.util.Date],arguments=[1, Sat Jul 29 15:46:46 CST 2017],attachments={path=com.wzf.funny.service.ArticleService,interface=com.wzf.funny.service.ArticleService, version=dev, timeout=500000}]]
     

    六、   consumer端返回结果调用链

    1.  处理从channel中获取的数据,执行Received方法

    1)    SimpleChannelHandler#handleUpstream

    位于netty.jar中,是response的入口

           if (e instanceofMessageEvent) {
               messageReceived(ctx,(MessageEvent) e);
           } 

    2)    NettyHandler#messageReceived

    获取NettyChannel并执行received方法

            NettyChannel channel =NettyChannel.getOrAddChannel(ctx.getChannel(),url, handler);
           try {
               handler.received(channel, e.getMessage());
           } finally {
               NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
           }



    3)    AbstractPeer#received

    先判断channel是否关闭,然后直接调用HeartbeatHandler#received方法

           setReadTimestamp(channel);
           if (isHeartbeatRequest(message)) {
               Request req= (Request) message;
               if (req.isTwoWay()) {
                   Response res = newResponse(req.getId(), req.getVersion());
                   res.setEvent(Response.HEARTBEAT_EVENT);
                   channel.send(res);
               }
               return;
           }
           if(isHeartbeatResponse(message)) {
               return;
           }
           handler.received(channel, message);
       }
    



    4)    HeartbeatHandler#received

    判断是否与心跳相关的,如果不是调用MultiMessageHandler#received方法。

           setReadTimestamp(channel);
           if (isHeartbeatRequest(message)) {
               Request req= (Request) message;
               if (req.isTwoWay()) {
                   Response res = newResponse(req.getId(), req.getVersion());
                   res.setEvent(Response.HEARTBEAT_EVENT);
                   channel.send(res);
               }
               return;
           }
           if(isHeartbeatResponse(message)) {
               return;
           }
           handler.received(channel, message);
       }
    

     

    5)    MultiMessageHandler#received

    如果是MultiMessage则循环调用AllChannelHandler#received;如果不是直接调用AllChannelHandler#received

           if (message instanceofMultiMessage) {
               MultiMessage list= (MultiMessage)message;
               for(Objectobj : list){
                   handler.received(channel, obj);
               }
           } else {
               handler.received(channel, message);
           }


     

    1)   AllChannelHandler#received

    先取得一个线程池,然后执行接收消息的线程ChannelEventRunnable。

          ExecutorService cexecutor =getExecutorService();
           try {
               cexecutor.execute(new ChannelEventRunnable(channel,handler, ChannelState.RECEIVED, message));
           } catch (Throwable t) {
               throw new ExecutionException(message,channel, getClass() + " error when process received event .", t);
           } 

    6)    ChannelEventRunnable#run

    根据ChanaelState不同,进入不同的处理逻辑。

     

              switch (state) {
               case CONNECTED:
                  handler.connected(channel);
                   break;
               case DISCONNECTED:
                  handler.disconnected(channel);
                   break;
               case SENT:
                  handler.sent(channel,message);
                   break;
               case RECEIVED:
                   handler.received(channel, message);
                   break;
               case CAUGHT:
                   handler.caught(channel, exception);
                   break;
               default:
                   logger.warn("unknown state: " + state + ",message is " + message);


    7)    DecodeHandler#received

    从message中获取Result,并decode;然后调用HeaderExchangeHandler#received方法。

           if (message instanceofDecodeable) {
               decode(message);
           }
           if (message instanceofRequest) {
               decode(((Request)message).getData());
           }
           if (message instanceofResponse) {
               decode( ((Response)message).getResult());
           }
           handler.received(channel, message);

    8)    HeaderExchangeHandler#received

    根据message的类型,进入不同处理逻辑,这里会进入handleResponse方法。

           channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
           ExchangeChannel exchangeChannel= HeaderExchangeChannel.getOrAddChannel(channel);
           try {
               if (message instanceofRequest) {
                    // handlerequest.
               } else if (message instanceof Response) {
                   handleResponse(channel, (Response) message);
               } else if (message instanceof String) {
                    // handle string.
               } else {
                   handler.received(exchangeChannel, message);
               }
           } finally {
              HeaderExchangeChannel.removeChannelIfDisconnected(channel);
           }
    

    9)    HeaderExchangeHandler#handleResponse

    如果不是心跳消息,那么通过DefaultFuture.received来接收消息。

            if (response != null&& !response.isHeartbeat()) {
               DefaultFuture.received(channel, response);
           } 

    2.  唤醒await的内容,继续之前的调用执行

    1)    DefaultFuture#received

    从FUTURES中(一个ConcurrentHasMap)根据删除这个DafaultFutrue,并调用DefaultFutrue#doReceived方法。

      try {
          DefaultFuture future = FUTURES.remove(response.getId());
          if (future != null){
              future.doReceived(response);
          } else {
               logger.warn("Thetimeout response finally returned at 。。。。。" );
          }
      } finally {
          CHANNELS.remove(response.getId());
      }


     

    2)    DefaultFuture#doReceived

    先上锁,然后唤醒之前await的内容。

           lock.lock();
           try {
               response= res;
               if (done != null) {
                   done.signal();
               }
           } finally {
               lock.unlock();
           }
           if (callback != null){
               invokeCallback(callback);
           }

     

    1)   DefaultFuture#get

    被唤醒后跳出while循环,调用returnFromResponse方法,拿到返回结果以后就可以继续之前DubboInvoker#doInvoke的调用了。

           if (timeout <= 0) {
               timeout =Constants.DEFAULT_TIMEOUT;
           }
           if (! isDone()) {
               long start = System.currentTimeMillis();
               lock.lock();
               try {
                   while(! isDone()) {
                       done.await(timeout,TimeUnit.MILLISECONDS);
                       if (isDone() || System.currentTimeMillis() - start > timeout){
                           break;
                       }
                   }
               } catch(InterruptedException e) {
                   thrownew RuntimeException(e);
               } finally{
                   lock.unlock();
               }
               if (!isDone()) {
                   thrownew TimeoutException(sent > 0, channel,getTimeoutMessage(false));
               }
           }
           returnreturnFromResponse();
    

     

    3)    DefaultFuture#returnFromResponse

    判断返回结果:如果返回结果为空,则返回IllegalStateException;如果成功,则返回result信息;如果客户端/服务端超时,则返回TimeoutException;如果其他错误,返回RemotingException。

           Response res = response;
           if (res == null) {
               throw new IllegalStateException("responsecannot be null");
           }
           if (res.getStatus() == Response.OK) {
               return res.getResult();
           }
           if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus()== Response.SERVER_TIMEOUT) {
               throw new TimeoutException(res.getStatus()== Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
           }
           throw new RemotingException(channel,res.getErrorMessage());
     

    七、   Provider端响应Rpc请求

    1.  处理从channel中获取的数据,执行Received方法

    此过程和Consumer端收到返回结果后,处理返回结果的流程基本相同,唯一不同的地方是,在最后一步,进入的是HeaderExchangeHandler#handleRequest方法

     

    1)    HeaderExchangeHandler#handleRequest

    调用通过DubboProtocol#replay来处理rpc请求

           Response res = new Response(req.getId(),req.getVersion());
           Object msg = req.getData();
            // handle data.
           Object result = handler.reply(channel,msg);
           res.setStatus(Response.OK);
           res.setResult(result);
           return res;

    2.  获取invoker,执行invoke方法。

    1)    DubboProtocol#reply

    根据message获取invoker对象,然后执行invoke方法,此调用会先进入一个拦截器链。

               if (message instanceofInvocation) {
                   Invocation inv = (Invocation) message;
                   Invoker<?> invoker = getInvoker(channel,inv);
                    //如果是callback需要处理高版本调用低版本的问题
                  RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                   returninvoker.invoke(inv);
               }
               throw new RemotingException(channel,"Unsupported request: "); 

    3.  执行Filter链

    1)    Filter链

    l   com.alibaba.dubbo.rpc.filter.EchoFilter

    methodName中包括$echo时,直接返回结果。

    l   com.alibaba.dubbo.rpc.filter.ClassLoaderFilter

    invoke前将ContextClassLoader设置为接口的ClassLoader,调用结束后将ContextClassLoader为当前线程的ContextClassLoader。

    l   com.alibaba.dubbo.rpc.filter.GenericFilter

    主要是针对泛化接口的实现。

    l   com.alibaba.dubbo.rpc.filter.ContextFilter

    对RpcContext进行赋值。

    l   com.alibaba.dubbo.rpc.protocol.dubbo.filter.TraceFilter

    方法调用的跟踪

    l   com.alibaba.dubbo.monitor.support.MonitorFilter

    监控rpc调用情况。

    l   com.alibaba.dubbo.rpc.filter.TimeoutFilter

    超时后,只是留了一个日志。

    l   com.alibaba.dubbo.rpc.filter.ExceptionFilter

    异常处理

     

    4.  Invoker

    1)    InvokerWrapper#invoke

    无业务逻辑,只是一层封装。

       public Result invoke(Invocation invocation) throwsRpcException {
           return invoker.invoke(invocation);
       }

    2)    AbstractProxyInvoker#invoke

    封装返回结果。

     public Result invoke(Invocation invocation) throwsRpcException {
               return new RpcResult(doInvoke(proxy,invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
       } 

    3)    JavassistProxyFactory#getInvoker

    在之前调用getInvoier的时候,会创建一个内部类,AbstractProxyInvoker# doInvoke方法会触发次内部内的执行。

    其中proxy就是Provider中定义的ref,即rpcServiceImpl

    另外除了JavassistProxyFactory以外,还有一个JdkProxyFactory。

       public <T> Invoker<T>getInvoker(T proxy, Class<T> type, URL url){
            // TODO Wrapper类不能正确处理带$的类名
           final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass(): type);
           return new AbstractProxyInvoker<T>(proxy, type, url) {
               @Override
               protectedObject doInvoke(T proxy, String methodName,   Class<?>[] parameterTypes, 
                    Object[] arguments) throwsThrowable {
                    return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
               }
           };
       }
    

     

     

     

    展开全文
  • 在Crystal框架下,如何开发基于Dubbo RPC的服务? 在Crystal框架下,如何调用Dubbo RPC服务? 相关的文章 什么是DubboDubbo[]是一个分布式服务框架,致力于提供高性能和透明化的RPC远程...
  • dubbo rpc Filter 中 invoker

    2018-05-23 00:39:00
    dubbo rpc Filter 中 invoker的具体原理 Builder构造器的使用 //TODO 转载于:https://my.oschina.net/u/3299170/bl...
  • 阿里在2017年重启dubbo框架的开发,2018年初合并当当网的dubbox分支,然后进入Apache基金会孵化, ...Apachedubbo RPC框架 中文文档:http://dubbo.apache.org/zh-cn/docs/user/quick-start.html ...
  • 基于dubbo RPC的微服务式架构 Dubbo Hello World 环境 SpringBoot + dubbo Pom.xml 依赖 <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent...
  • <div><p>是否支持dubbo rpc隐式参数传递呢?如何支持</p><p>该提问来源于开源项目:thubbo/jmeter-plugins-for-apache-dubbo</p></div>
  • 一、为什么要集成Dubbo RPC 在国内建设微服务有Dubbo和Spring Cloud两种主流方案可供选择。Dubbo是服务治理与RPC实现方案(可能理解为一个分布式通信框架),虽然在国内有着非常大的用户群体,但是其周边设施与...
  • dubbo RPC框架原理解析

    2019-01-09 09:01:00
    dubbo是一个rpc框架,面向服务治理,是一款优秀的框架。
  • 分布式系统的跟踪系统Dubbo RPC处理

    千次阅读 2015-10-10 11:03:16
    分布式系统的跟踪系统Dubbo RPC处理  接着前一篇博文http://blog.csdn.net/doctor_who2004/article/details/46974695   上篇只是提供了一个思想,今天具体给出dubbo rpc 处理细节。  dubbo prc处理部分,那...
  • dubbo rpc框架

    2020-04-16 18:08:03
    dubbo应用架构
  • dubbo rpc 异常

    千次阅读 2018-03-27 15:04:10
    Caused by: com.alibaba.dubbo.rpc.RpcException: Failed to invoke the method subscribe in the service com.alibaba.dubbo.registry.RegistryService. Tried 3 times of the providers[172.168.1.167:2181] (1/1...
  • SpringBoot Zipkin Dubbo Rpc Http日志链路追踪全流程 先拍砖,这个流程你跑不通,或者说Dubbo调用时候服务提供者、消费者 traceId不能串联起来、直接来拍我,有一个我接一个!!!!!! 本次主要针对 2.7.x Dubbo...
  • Dubbo RPC协议 Dubbo目前对dubbo、rmi、hessian、http、webservice、thrift、memecached、redis、rest多种协议的支持,框架在默认情况下支持dubbo协议。下面我们讲主要对前四种协议进行详细的讲解以及实战的压测。 ...
  • SpringBoot Zipkin Dubbo Rpc Http日志链路追踪全流程 先拍砖,这个流程你跑不通,或者说Dubbo调用时候服务提供者、消费者 traceId不能串联起来、直接来拍我,有一个我接一个!!!!!! 说明: 本次主要针对 2.7.x...
  • dubbo rpc接口文档生成

    千次阅读 2020-07-05 23:06:22
    在国内dubbo成为很多互联网公司高并发分布式场景下rpc框架的首选,dubbo从开源至今经历过蛮多的过程,从开源到中间的停止维护,经过三年的沉寂,2017年9月,阿里巴巴宣布重启dubbo项目。到2018年2月,阿里将dubbo...
  • 第 9 章 实现简易版 dubbo RPC 一、RPC 基本介绍 RPC(RemoteProcedure Call)— 远程过程调用,是一个计算机通信协议。该协议允许运行于一台计算机的程 序调用另一台计算机的子程序,而程序员无需额外地为这个交互...
  • 作者:hu1991die | 整理:根哥啊 ... RPC非常重要,很多人面试的时候都挂在了这个地方!...特别是文中给出的一张关于RPC的基本流程图,重点中的重点,Dubbo RPC的基本执行流程就是他,RPC框架的基本原理也是他,别说...
  • Dubbo RPC 基础

    2019-05-05 00:03:26
  • 使用Netty实现Dubbo RPC

    千次阅读 2020-02-15 15:01:45
    RPC基本介绍 1.远程过程调用,是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序,而程序员无需额外地为这个交互...4.常用的PRC 框架:比较知名的 Dubbo,gRPC,Spring Cloud 基于ne...
  • Spring Cloud Alibaba Dubbo RPC 2.7.x & Nacos 1.4.x & spring Boot 2.4.x 工程使用 1.RPC和HTTP RPC主要用于公司内部的服务调用,性能消耗低,传输效率高,服务治理方便。 HTTP主要用于对外的异构环境, ...
  • 关注 “Java艺术” 我们一起成长!试下这篇能不能把去年写的Dubbo源码分析系列文章重新激活Dubbo RPC使用dubbo协议只需要一个长连接就可以收发所有请求,为什么使用http...
  • 我在使用dubbo RPC请求时,参数中包含泛型的List集合对象,使用post方式提交时,系统后台报错如下 user-agent: Mozilla/5.0 (Windows NT 6.1; WOW64; rv:53.0) Gecko/20100101 Firefox/53.0 , dubbo version: 2.8.4,...

空空如也

空空如也

1 2 3 4 5 ... 20
收藏数 13,078
精华内容 5,231
关键字:

dubborpc