投稿
  1. 首页
  2. 随笔

Dubbo源码解析-Dubbo服务消费者

美丽教师来我家 • 2022-12-15 • 随笔 • 阅读 11

Dubbo源码解析-Dubbo服务消费者,第1张

Dubbo源码解析-Dubbo服务消费者 前言：

    前两篇我们分析了Dubbo服务提供者，在创建时的基本配置属性，如ServiceConfig、ApplicationConfig、RegistryConfig等。基本了解了基于API方式来创建Dubbo服务提供者的套路。

    同时第二篇我们分析了在injvm（本地）模式下，dubbo服务如何向外注册（本质上还是注册在本地的map中，就是InjvmProtocol.exporterMap中），等待同进程中的服务消费者来调用（这个很重要，笔者基于Dubbo-demo源码中测试代码进行测试时候，无论怎样，消费者都会报错，找不到服务提供者，就是因为消费者和提供者不在同一个进程内）。

    下面我们就来看下基于Injvm模式下的服务消费者创建方式及源码分析。

1.Injvm模式下的服务消费者和创建者

   代码来自dubbo-2.7.7中dubbo-demo-api项目下的Application，笔者有所精简（最主要的是将服务的发布和消费放到同一个main方法中了）

public class Application {
	// 服务提供者代码有所精简，本质上还是与之前的示例一样
	public static void main(String[] args) throws Exception {
        startWithExport();
        runWithRefer();
    }

    // 服务提供者
    private static void startWithExport() throws InterruptedException {
        ServiceConfig service = new ServiceConfig<>();
        service.setInterface(DemoService.class);
        service.setRef(new DemoServiceImpl());
        service.setApplication(new ApplicationConfig("dubbo-demo-api-provider"));
        service.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));

        // 只暴露在本地，当前进程内
        service.setScope("local");
        service.export();

        System.out.println("dubbo service started");
//        new CountDownLatch(1).await();
    }

	// 服务消费者
    private static void runWithRefer() {
        ReferenceConfig reference = new ReferenceConfig<>();
        reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
        reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
        reference.setInterface(DemoService.class);
        // 寻找本地模式的服务提供者
        reference.setScope("local");
        DemoService service = reference.get();
        String message = service.sayHello("dubbo");
        System.out.println(message);
    }
}

有关于ApplicationConfig、RegistryConfig等之前都已经有过介绍，不再赘述。我们主要来看下ReferenceConfig

1.1 ReferenceConfig 
public class ReferenceConfig extends ReferenceConfigbase {
 	// 在 Dubbo服务提供（Injvm）中我们有过分析，这里返回的就是Protocol$Adaptive，
    // 调用链是Protocol$Adaptive --> ProtocolFilterWrapper  --> ProtocolListenerWrapper --> InjvmProtocol
    private static final Protocol REF_PROTOCOL = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
	// 这个暂时用不上，可以先忽略
    private static final Cluster CLUSTER = ExtensionLoader.getExtensionLoader(Cluster.class).getAdaptiveExtension();
	// 同理，这里返回的是ProxyFactory$Adaptive，调用链是StubProxyFactoryWrapper --> JavassistProxyFactory
    private static final ProxyFactory PROXY_FACTORY = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

    // 注册的接口类
    private transient volatile T ref;
    // ReferenceConfig主要属性就是以上
    ...
}

// ReferenceConfigbase
public abstract class ReferenceConfigbase extends AbstractReferenceConfig {
 
    // 接口名
    protected String interfaceName;

    //接口类信息
    protected Class interfaceClass;

    
    protected String client;

    // 用于点对点模式下指定调用的url
    protected String url;

    // 消费者配置信息（指定线程数等信息）
    protected ConsumerConfig consumer;

    // 指定协议
    protected String protocol;

    protected Servicemetadata servicemetadata;
}

主要属性就是上述这些，没有什么特别的东西。

有关于Protocol$Adaptive、ProxyFactory$Adaptive的内容可参考  Dubbo源码解析-Dubbo服务提供者_Injvm协议（二）_恐龙弟旺仔的博客-CSDN博客

2.Injvm消费者源码分析 2.1 ReferenceConfig.get() 
public class ReferenceConfig extends ReferenceConfigbase {
	public synchronized T get() {
        if (destroyed) {
            throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
        }
        // 交由init()方法处理
        if (ref == null) {
            init();
        }
        return ref;
    }
    
    public synchronized void init() {
        // 初始化过的则不再初始化
        if (initialized) {
            return;
        }
		...
		// 这里主要用于设置属性 servicemetadata
        checkAndUpdateSubConfigs();

		...
        Map map = new HashMap();
        map.put(SIDE_KEY, CONSUMER_SIDE);

        ReferenceConfigbase.appendRuntimeParameters(map);
        ...
        // 一系列的配置参数，不是重点，直接忽略
        map.put(INTERFACE_KEY, interfaceName);
        AbstractConfig.appendParameters(map, getMetrics());
        ...
        servicemetadata.getAttachments().putAll(map);

        // 这里是重点
        ref = createProxy(map);

        servicemetadata.setTarget(ref);
        servicemetadata.addAttribute(PROXY_CLASS_REF, ref);
        ...
    }
}

这里主要是解析ApplicationConfig等配置的参数到map中，最终添加到ServiceConfig中

// 在笔者当前的示例中，map的内容如下：
	"side" -> "consumer"
    "application" -> "dubbo-demo-api-consumer"
    "register.ip" -> "xxx.xxx"
    "release" -> ""
    "methods" -> "sayHello,sayHelloAsync"
    "scope" -> "local"
    "sticky" -> "false"
    "dubbo" -> "2.0.2"
    "pid" -> "9612"
    "interface" -> "org.apache.dubbo.demo.DemoService"
    "timestamp" -> "1628079039393"
2.2 ReferenceConfig.createProxy() 
public class ReferenceConfig extends ReferenceConfigbase {
	private T createProxy(Map map) {
        // 这里判断是否injvm模式，本例中时 是
        if (shouldJvmRefer(map)) {
            // 本例中生成的url 内容为：injvm://127.0.0.1/org.apache.dubbo.demo.DemoService?application=dubbo-demo-api-consumer&dubbo=2.0.2&interface=org.apache.dubbo.demo.DemoService&methods=sayHello,sayHelloAsync&pid=9612®ister.ip=172.20.89.171&release=&scope=local&side=consumer&sticky=false×tamp=1628079039393
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            
            // 这里通过代理对象生成invoker，通过之前的分析我们知道，最终会调用InjvmProtocol.refer()方法，具体分析见2.2.1
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
            if (logger.isInfoEnabled()) {
                logger.info("Using injvm service " + interfaceClass.getName());
            }
        } else {
            // 非本地模式的不是本文重点，先忽略
        }

        // 如果当前没有设置check=false，并且Invoker没有找到合适的服务提供者（怎么找呢，见2.2.2）
        if (shouldCheck() && !invoker.isAvailable()) {
            invoker.destroy();
            throw new IllegalStateException("Failed to check the status of the service "
                    + interfaceName
                    + ". No provider available for the service "
                    + (group == null ? "" : group + "/")
                    + interfaceName +
                    (version == null ? "" : ":" + version)
                    + " from the url "
                    + invoker.getUrl()
                    + " to the consumer "
                    + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
        }
        ...
        String metadata = map.get(metaDATA_KEY);
        WritablemetadataService metadataService = WritablemetadataService.getExtension(metadata == null ? DEFAULT_metaDATA_STORAGE_TYPE : metadata);
        if (metadataService != null) {
            URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
            metadataService.publishServiceDefinition(consumerURL);
        }
        // 重要的又来了，我们接着看
        return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    }
}

2.2.1 Protocol$Adaptive.refer()

调用链依旧是 Protocol$Adaptive --> ProtocolFilterWrapper  --> ProtocolListenerWrapper --> InjvmProtocol

我们直接看InjvmProtocol.refer()方法

public abstract class AbstractProtocol implements Protocol {
 
    public  Invoker refer(Class type, URL url) throws RpcException {
        // 最终由InjvmProtocol.protocolBindingRefer实现
        return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
    }
}

// InjvmProtocol.protocolBindingRefer
public class InjvmProtocol extends AbstractProtocol implements Protocol {
    public  Invoker protocolBindingRefer(Class serviceType, URL url) throws RpcException {
        // 这个exporterMap很重要，就是之前我们injvm模式下，服务提供者将当前服务注册的地方，在当前Invoker中，直接把这个map当做构造参数传入
        return new InjvmInvoker(serviceType, url, url.getServiceKey(), exporterMap);
    }
}

class InjvmInvoker extends AbstractInvoker {
    private final Map> exporterMap;
    
    InjvmInvoker(Class type, URL url, String key, Map> exporterMap) {
        super(type, url);
        this.key = key;
        this.exporterMap = exporterMap;
    }
}

重要：我们要正确的理解exporterMap，服务提供者（Injvm模式）会将当前服务添加到这个map中，key就是我们当前的接口全限定名。

现在我们在创建服务消费者（Injvm模式），会将InjvmProtocol中的这个exporterMap当做构造参数直接传入InjvmInvoker中，正好作为后续的使用。

2.2.2 invoker.isAvailable()

    本质上还是InjvmInvoker.isAvailable()方法的调用

class InjvmInvoker extends AbstractInvoker {
    public boolean isAvailable() {
        // 接着上面的看，就是看exporterMap是否有当前接口的服务提供者
        InjvmExporter exporter = (InjvmExporter) exporterMap.get(key);
        if (exporter == null) {
            return false;
        } else {
            return super.isAvailable();
        }
    }
}
2.3 PROXY_FACTORY.getProxy()

    正确的调用链就是 ProxyFactory$Adaptive --> StubProxyFactoryWrapper --> JavassistProxyFactory

    我们直接看JavassistProxyFactory.getProxy()

public class JavassistProxyFactory extends AbstractProxyFactory {
    public  T getProxy(Invoker invoker, Class[] interfaces) {
        // 之前的文章中有分析过Proxy.getProxy，就是使用javassist动态创建代理类，最终还是调用InjvmInvoker来实现
        return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    }
}

// Proxy.getProxy()
public abstract class Proxy {
	public static Proxy getProxy(ClassLoader cl, Class... ics) {
        ...

            // create ProxyInstance class.
            String pcn = pkg + ".proxy" + id;
            ccp.setClassName(pcn);
            ccp.addField("public static java.lang.reflect.Method[] methods;");
            ccp.addField("private " + InvocationHandler.class.getName() + " handler;");
            ccp.addConstructor(Modifier.PUBLIC, new Class[]{InvocationHandler.class}, new Class[0], "handler=$1;");
            ccp.addDefaultConstructor();
            Class clazz = ccp.toClass();
            clazz.getField("methods").set(null, methods.toArray(new Method[0]));

            // create Proxy class.
            String fcn = Proxy.class.getName() + id;
            ccm = ClassGenerator.newInstance(cl);
            ccm.setClassName(fcn);
            ccm.addDefaultConstructor();
            ccm.setSuperClass(Proxy.class);
            ccm.addMethod("public Object newInstance(" + InvocationHandler.class.getName() + " h){ return new " + pcn + "($1); }");
            Class pc = ccm.toClass();
        // 还是动态生成代理那一套，我们直接看下动态生成的代理类的内容，见2.3.1
            proxy = (Proxy) pc.newInstance();
        } catch (RuntimeException e) {
            ...
        }
        return proxy;
    }
}

2.3.1 org.apache.dubbo.common.bytecode.Proxy0

    如何查看动态生成的类呢？可以参考这篇文章：https://blog.csdn.net/yfxhao123/article/details/109472166 

package org.apache.dubbo.common.bytecode;
import java.lang.reflect.InvocationHandler;

// 其继承的这个Proxy就是org.apache.dubbo.common.bytecode.Proxy抽象类，并实现了其抽象方法newInstance()
public class Proxy0
  extends Proxy
  implements ClassGenerator.DC
{
  public Object newInstance(InvocationHandler paramInvocationHandler)
  {
    return new proxy0(paramInvocationHandler);
  }
}

 

总结：我们上面示例代码中reference.get()最终返回的就是一个调用代理类Proxy（针对InjvmInvoker的代理类InvokerInvocationHandler对象）

最终，我们还是要通过InvokerInvocationHandler来真正实现调用。

2.4 InvokerInvocationHandler.invoke() 
public class InvokerInvocationHandler implements InvocationHandler {
 
    // 这里传入的proxy就是 上面的InjvmInvoker对象
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(invoker, args);
        }
        String methodName = method.getName();
        Class[] parameterTypes = method.getParameterTypes();
        ...
        // 创建RpcInvocation
        RpcInvocation rpcInvocation = new RpcInvocation(method, invoker.getInterface().getName(), args);
        String serviceKey = invoker.getUrl().getServiceKey();
        rpcInvocation.setTargetServiceUniqueName(serviceKey);
      
        if (consumerModel != null) {
            rpcInvocation.put(Constants.CONSUMER_MODEL, consumerModel);
            rpcInvocation.put(Constants.METHOD_MODEL, consumerModel.getMethodModel(method));
        }

        // 还是交由invoker执行，后续的还是熟悉的配方，调用链依次执行ProtocolFilterWrapper --> ListenerInvokerWrapper --> InjvmInvoker
        return invoker.invoke(rpcInvocation).recreate();
    }
}

中间的ProtocolFilterWrapper、ListenerInvokerWrapper，我们先不分析，后续讲到Filter时会仔细说明。

直接来看最终的调用InjvmInvoker

2.4 InjvmInvoker.doInvoker() 
class InjvmInvoker extends AbstractInvoker {
	public Result doInvoke(Invocation invocation) throws Throwable {
        // 获取之前存放在exporterMap中的InjvmExporter
        Exporter exporter = InjvmProtocol.getExporter(exporterMap, getUrl());
        if (exporter == null) {
            throw new RpcException("Service [" + key + "] not found.");
        }
        RpcContext.getContext().setRemoteAddress(LOCALHOST_VALUE, 0);
        
        // InjvmExport.invoker就是我们之前分析的，忘记的可以再翻翻看下 Dubbo服务提供（Injvm）
        return exporter.getInvoker().invoke(invocation);
    }
}

貌似又回来了，我们拿到服务提供者的InjvmExporter之后，其Invoker属性就是之前文章中说的 JavassistProxyFactory.getInvoker()方法所返回的那个AbstractProxyInvoker

2.5 调回到AbstractProxyInvoker.doInvoke() 
public class JavassistProxyFactory extends AbstractProxyFactory {
	...
    @Override
    public  Invoker getInvoker(T proxy, Class type, URL url) {
        // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
        final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
        return new AbstractProxyInvoker(proxy, type, url) {
            @Override
            protected Object doInvoke(T proxy, String methodName,
                                      Class[] parameterTypes,
                                      Object[] arguments) throws Throwable {
                // 最终执行到这个方法，而wrapper.invokeMethod()的调用见2.5.1
                return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
            }
        };
    }
}

2.5.1 Wrapper1.invokeMethod()

public class Wrapper0
  extends Wrapper
  implements ClassGenerator.DC
{
public Object invokeMethod(Object paramObject, String paramString, Class[] paramArrayOfClass, Object[] paramArrayOfObject)
    throws InvocationTargetException
  {
    DemoServiceImpl localDemoServiceImpl;
    try
    {
      localDemoServiceImpl = (DemoServiceImpl)paramObject;
    }
    catch (Throwable localThrowable1)
    {
      throw new IllegalArgumentException(localThrowable1);
    }
    try
    {
    // 在这里实现了对localDemoServiceImpl的调用，就是我们在创建ServiceConfig时setRef()方法所引用的类，本例中即为 new DemoServiceImpl()
      if ((!"sayHello".equals(paramString)) || (paramArrayOfClass.length == 1)) {
        return localDemoServiceImpl.sayHello((String)paramArrayOfObject[0]);
      }
      if ((!"sayHelloAsync".equals(paramString)) || (paramArrayOfClass.length == 1)) {
        return localDemoServiceImpl.sayHelloAsync((String)paramArrayOfObject[0]);
      }
    }
    catch (Throwable localThrowable2)
    {
      throw new InvocationTargetException(localThrowable2);
    }
    throw new NoSuchMethodException("Not found method "" + paramString + "" in class org.apache.dubbo.demo.provider.DemoServiceImpl.");
  }
}
总结：

    1.通过创建ReferenceConfig，在调用其get()方法时，会返回InjvmInvoker对象，该对象的exporterMap即当前InjvmProtocol下的所有服务提供者信息；

    2.通过ProxyFactory$Adaptive创建对InjvmInvoker的代理，最终调用到JavassistProxyFactory.getProxy()来实现，动态创建了对InjvmInvoker的代理类Proxy0；

    3.在调用业务方法Proxy0.sayHello()时，实际调用的是InjvmInvoker.doInvoker()方法，这里会从之前保存的exporterMap中获取对应的Exporter实现类，并获取Exporter.getInvoker()对象，调用其invoke()方法，此时我们又回到了服务端的逻辑

    4.Exporter.getInvoker()本质上就是JavassistProxyFactory.getInvoker()方法返回值，即AbstractProxyInvoker，其doInvoke()调用交由wrapper实现

    5.根据我们对Wrapper1（动态生成的对象）进行解析，可以发现，其最终调用的还是我们在ServiceConfig.setRef()中设置的实现类。

最重要的两个方法：

public class ReferenceConfig extends ReferenceConfigbase {
	private T createProxy(Map map) {
        if (shouldJvmRefer(map)) {
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            // 1.生成Invoker
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
        } else {
            // 非本地模式的不是本文重点，先忽略
        }

        ...
        // 2.生成Proxy
        return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    }
}

所以，服务消费者的主要过程也就是：将消费转变为Invoker（本例中为InjvmInvoker）；将Invoker转变为Proxy

最后，用一张时序图来展示下消费者（Injvm模式）全过程

 

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5606822.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
美丽教师来我家 美丽教师来我家 一级用户组
0 0
上一篇 2022-12-15
下一篇 2022-12-15

发表评论

登录后才能评论

评论列表（0条）

保存