当前位置： 代码迷 >> 综合 >> dubbo使用9----＞ dubbo 消费者订阅流程源码解析
  详细解决方案

dubbo使用9----＞ dubbo 消费者订阅流程源码解析

热度：62   发布时间：2023-10-24 15:06:50.0

1、dubbo服务引用方式

      1.1、基于xml的方式：

    <dubbo:reference interface="com.wzy.api.ISayHelloService" id="sayHelloService" />

      1. 2、基于注解：

    @DubboReference(check = false)

      1.3、基于代码：此处不再累赘，官网有案例。

 

2、dubbo服务引用原理分析

      步骤1 ：ReferenceBean.get()方法：ReferenceBean实现了FactoryBean接口，因此来到实现的getObject()方法

    @Overridepublic Object getObject() {return get();}

      步骤2：ReferenceBean父类ReferenceConfig的 get()方法：

    public synchronized T get() {if (destroyed) {throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");}如果需要注入的dubbo引用为空，那就初始化后再返回refif (ref == null) {init();}return ref;}

      步骤3：ReferenceConfig的init()方法：

    public synchronized void init() {1、如果已经初始化过了，结束方法if (initialized) {return;}2、先初始化DubboBootstrap，主要操作是检查配置，加载一些配置。if (bootstrap == null) {bootstrap = DubboBootstrap.getInstance();bootstrap.init();}checkAndUpdateSubConfigs();checkStubAndLocal(interfaceClass);ConfigValidationUtils.checkMock(interfaceClass, this);3、创建一个map用于存放配置的参数Map<String, String> map = new HashMap<String, String>();map.put(SIDE_KEY, CONSUMER_SIDE);ReferenceConfigBase.appendRuntimeParameters(map);if (!ProtocolUtils.isGeneric(generic)) {String revision = Version.getVersion(interfaceClass, version);if (revision != null && revision.length() > 0) {map.put(REVISION_KEY, revision);}String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();if (methods.length == 0) {logger.warn("No method found in service interface " + interfaceClass.getName());map.put(METHODS_KEY, ANY_VALUE);} else {map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR));}}map.put(INTERFACE_KEY, interfaceName);AbstractConfig.appendParameters(map, getMetrics());AbstractConfig.appendParameters(map, getApplication());AbstractConfig.appendParameters(map, getModule());// remove 'default.' prefix for configs from ConsumerConfig// appendParameters(map, consumer, Constants.DEFAULT_KEY);AbstractConfig.appendParameters(map, consumer);AbstractConfig.appendParameters(map, this);MetadataReportConfig metadataReportConfig = getMetadataReportConfig();if (metadataReportConfig != null && metadataReportConfig.isValid()) {map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);}Map<String, AsyncMethodInfo> attributes = null;if (CollectionUtils.isNotEmpty(getMethods())) {attributes = new HashMap<>();for (MethodConfig methodConfig : getMethods()) {AbstractConfig.appendParameters(map, methodConfig, methodConfig.getName());String retryKey = methodConfig.getName() + ".retry";if (map.containsKey(retryKey)) {String retryValue = map.remove(retryKey);if ("false".equals(retryValue)) {map.put(methodConfig.getName() + ".retries", "0");}}AsyncMethodInfo asyncMethodInfo = AbstractConfig.convertMethodConfig2AsyncInfo(methodConfig);if (asyncMethodInfo != null) {
//                    consumerModel.getMethodModel(methodConfig.getName()).addAttribute(ASYNC_KEY, asyncMethodInfo);attributes.put(methodConfig.getName(), asyncMethodInfo);}}}String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);if (StringUtils.isEmpty(hostToRegistry)) {hostToRegistry = NetUtils.getLocalHost();} else if (isInvalidLocalHost(hostToRegistry)) {throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);}map.put(REGISTER_IP_KEY, hostToRegistry);serviceMetadata.getAttachments().putAll(map);4、使用存放配置参数的map去创建一个代理类，然后赋值给ref属性，那就表示依赖注入的实例就是一个代理类。ref = createProxy(map);serviceMetadata.setTarget(ref);serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey());consumerModel.setProxyObject(ref);consumerModel.init(attributes);5、修改初始化变量为true表示当前的服务引用以及初始化完成。initialized = true;checkInvokerAvailable();// dispatch a ReferenceConfigInitializedEvent since 2.7.4dispatch(new ReferenceConfigInitializedEvent(this, invoker));}

       步骤4：ReferenceConfig的createProxy(Map<String, String> map)方法：根据配置的参数去构建一个代理类，此处是核心中的核心；

    private T createProxy(Map<String, String> map) {if (shouldJvmRefer(map)) {1、如果需要发布injvm协议那就发布。URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);invoker = REF_PROTOCOL.refer(interfaceClass, url);if (logger.isInfoEnabled()) {logger.info("Using injvm service " + interfaceClass.getName());}} else {urls.clear();2、如果是直连，那就进行url处理。if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.String[] us = SEMICOLON_SPLIT_PATTERN.split(url);if (us != null && us.length > 0) {for (String u : us) {URL url = URL.valueOf(u);if (StringUtils.isEmpty(url.getPath())) {url = url.setPath(interfaceName);}if (UrlUtils.isRegistry(url)) {urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));} else {urls.add(ClusterUtils.mergeUrl(url, map));}}}} else { // assemble URL from register center's configuration// if protocols not injvm checkRegistry3、如果不需要发布injvm协议，那就载入注册中心的地址。if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {checkRegistry();List<URL> us = ConfigValidationUtils.loadRegistries(this, false);if (CollectionUtils.isNotEmpty(us)) {for (URL u : us) {URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);if (monitorUrl != null) {map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));}urls.add(u.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));}}if (urls.isEmpty()) {throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");}}}if (urls.size() == 1) {4、如果是单注册中心那就使用注册中心的地址来进行引用，会使用到一个自适应扩展点Protocol，而此时由于url是注册中心的地址因此此处的协议protocol应该是被包装的 RegistryProtocol实例。invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0));} else {5、如果是多注册中心那就进行循环引用，我们以单注册中心为例进行源码分析，多注册中心大同小异。List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();URL registryURL = null;for (URL url : urls) {invokers.add(REF_PROTOCOL.refer(interfaceClass, url));if (UrlUtils.isRegistry(url)) {registryURL = url; // use last registry url}}if (registryURL != null) { // registry url is available// for multi-subscription scenario, use 'zone-aware' policy by defaultString cluster = registryURL.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);// The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invokerinvoker = Cluster.getCluster(cluster, false).join(new StaticDirectory(registryURL, invokers));} else { // not a registry url, must be direct invoke.String cluster = CollectionUtils.isNotEmpty(invokers)? (invokers.get(0).getUrl() != null ? invokers.get(0).getUrl().getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME) : Cluster.DEFAULT): Cluster.DEFAULT;invoker = Cluster.getCluster(cluster).join(new StaticDirectory(invokers));}}}6、打印成功引用日志。if (logger.isInfoEnabled()) {logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());}/*** @since 2.7.0* ServiceData Store*/7、将服务的元数据进行保存，默认会写入到本地，可进行metadata=remote进行配置。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);}// create service proxy8、最后使用服务引用生成的invoker来构建一个代理类，默认是使用javassist技术来构建。return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));}

    步骤5：Protocol的refer(Class<T> type, URL url)方法：type是接口类型、url是载入后的注册中心地址；

                 Protocol是一个自适应扩展点，而此时的url的协议是registry协议，例如registry://xxxx因此此时的自适应扩展其实就是被包装的RegistryProtocol实现，包装的顺序如下：第一层是QosProtocolWrapper

    步骤6：RegistryProtocol的refer(Class<T> type, URL url)方法：type是接口类型、url是载入后的注册中心地址；

    @Override@SuppressWarnings("unchecked")public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {1、先将注册中心地址url转换为真正协议的注册中心地址如zookeeper://xxxurl = getRegistryUrl(url);2、获取一个注册器，可用于注册、卸载、订阅、通知等操作，比如zookeeperRegistry 在此步骤
会构建zkClient进行连接，然后注册watch事件。Registry registry = registryFactory.getRegistry(url);if (RegistryService.class.equals(type)) {return proxyFactory.getInvoker((T) registry, type, url);}// group="a,b" or group="*"Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));3、获取配置服务的group信息，如果是需要合并的那就进行合并引用。String group = qs.get(GROUP_KEY);if (group != null && group.length() > 0) {if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {return doRefer(Cluster.getCluster(MergeableCluster.NAME), registry, type, url);}}4、获取集群容错策略。Cluster cluster = Cluster.getCluster(qs.get(CLUSTER_KEY));5、进行服务引用。return doRefer(cluster, registry, type, url);}

    步骤7：RegistryProtocol的doRefer(Cluster cluster, Registry registry, Class<T> type, URL url)方法：cluster参数是集群荣容错策略，也是一个扩展点，所以实例当然是包装过后的；type是接口类型；url是载入后的注册中心地址，zookeeper：//xxx例如；

    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {1、为当前接口创建一个RegistryDirectory，可以理解为目录，当前接口的服务提供者的目录，
里面会维护Invoker列表，每一个invoker就是一个服务提供者节点封装，在订阅的时候会将服务提供者地址
从注册器上找到，然后转换成每一个Invoker实例存在directory中。RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);2、设置目录的注册器。directory.setRegistry(registry);3、设置目录的协议。directory.setProtocol(protocol);// all attributes of REFER_KEY4、构建一个消费地址例如consumer://192.168.0.113/com.wzy.dubbo.ISayHelloService?
application=consumer&check=false&dubbo=2.0.2&init=false&interface=com.wzy.dubbo.ISayHello
Service&metadatatype=remote&methods=hhhh,sayHello&pid=4220&qos.enable=false&release=2.7.8
&side=consumer&sticky=false&timestamp=1602340494748Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);5、当前的消费者是否需要注册，判断标准是消费端的register配置，默认是需要注册。if (directory.isShouldRegister()) {设置当前服务的注册目录的消费者urldirectory.setRegisteredConsumerUrl(subscribeUrl);将消费者地址注册到注册中心，如果是zookeeper的话就会进行消费者地址写入。registry.register(directory.getRegisteredConsumerUrl());}6、设置服务路由链，根据配置进行设置。directory.buildRouterChain(subscribeUrl);7、订阅当前服务的提供者。directory.subscribe(toSubscribeUrl(subscribeUrl));8、将当前的服务目录加入到集群中并返回一个InvokerInvoker<T> invoker = cluster.join(directory);List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);if (CollectionUtils.isEmpty(listeners)) {return invoker;}RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker);for (RegistryProtocolListener listener : listeners) {listener.onRefer(this, registryInvokerWrapper);}return registryInvokerWrapper;}订阅服务实现public void subscribe(URL url) {1、设置当前服务目录的消费者地址。setConsumerUrl(url);2、将当前RegistryDirectory实例添加到ConsumerConfigurationListener列表中，因为RegistryDirectory也实现了NotifyListener通知监听器，用于通知当前目录提供者的地址修改。CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this);3、设置服务配置的监听器。serviceConfigurationListener = new ReferenceConfigurationListener(this, url);4、使用注册器进行订阅，我们在下面以ZookeeperRegistry进行讲解。registry.subscribe(url, this);}

     我把核心步骤以画图的方式来完成服务引用过程：

 

 

 

 

 

 

 

 

 

 

 

 

 

 

  相关解决方案

代码迷 - 您身边的软件异常，代码异常，编程异常助手（发现，解决，分享）
Copyright © 2009-2013 DAIMAMI.COM. All rights reserved.