dubbo源码-服务发现

本是古典何须时尚 2021-11-22 13:44 391阅读 0赞

服务消费者

<dubbo:application name="dubbo-consumer"/>
    
        
        <dubbo:registry protocol="zookeeper" address="${dubbo.registry.address}"/>
    
        
        <dubbo:reference id="userService" check="false" interface="com.study.dubbo.userapi.service.UserService" loadbalance="consistenthash"/>

ReferenceBean

public class ReferenceBean<T> extends ReferenceConfig<T> implements FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {
    
        private static final long serialVersionUID = 213195494150089726L;
    
        private transient ApplicationContext applicationContext;
    
        public ReferenceBean() {
            super();
        }
    
        public ReferenceBean(Reference reference) {
            super(reference);
        }
    
        public void setApplicationContext(ApplicationContext applicationContext) {
            this.applicationContext = applicationContext;
            SpringExtensionFactory.addApplicationContext(applicationContext);
        }
    
        public Object getObject() throws Exception {
            return get();
        }
    
        public Class<?> getObjectType() {
            return getInterfaceClass();
        }
    
        @Parameter(excluded = true)
        public boolean isSingleton() {
            return true;
        }
    
        @SuppressWarnings({"unchecked"})
        public void afterPropertiesSet() throws Exception {
            if (getConsumer() == null) {
                Map<String, ConsumerConfig> consumerConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ConsumerConfig.class, false, false);
                if (consumerConfigMap != null && consumerConfigMap.size() > 0) {
                    ConsumerConfig consumerConfig = null;
                    for (ConsumerConfig config : consumerConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault().booleanValue()) {
                            if (consumerConfig != null) {
                                throw new IllegalStateException("Duplicate consumer configs: " + consumerConfig + " and " + config);
                            }
                            consumerConfig = config;
                        }
                    }
                    if (consumerConfig != null) {
                        setConsumer(consumerConfig);
                    }
                }
            }
            if (getApplication() == null
                    && (getConsumer() == null || getConsumer().getApplication() == null)) {
                Map<String, ApplicationConfig> applicationConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ApplicationConfig.class, false, false);
                if (applicationConfigMap != null && applicationConfigMap.size() > 0) {
                    ApplicationConfig applicationConfig = null;
                    for (ApplicationConfig config : applicationConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault().booleanValue()) {
                            if (applicationConfig != null) {
                                throw new IllegalStateException("Duplicate application configs: " + applicationConfig + " and " + config);
                            }
                            applicationConfig = config;
                        }
                    }
                    if (applicationConfig != null) {
                        setApplication(applicationConfig);
                    }
                }
            }
            if (getModule() == null
                    && (getConsumer() == null || getConsumer().getModule() == null)) {
                Map<String, ModuleConfig> moduleConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, ModuleConfig.class, false, false);
                if (moduleConfigMap != null && moduleConfigMap.size() > 0) {
                    ModuleConfig moduleConfig = null;
                    for (ModuleConfig config : moduleConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault().booleanValue()) {
                            if (moduleConfig != null) {
                                throw new IllegalStateException("Duplicate module configs: " + moduleConfig + " and " + config);
                            }
                            moduleConfig = config;
                        }
                    }
                    if (moduleConfig != null) {
                        setModule(moduleConfig);
                    }
                }
            }
            if ((getRegistries() == null || getRegistries().isEmpty())
                    && (getConsumer() == null || getConsumer().getRegistries() == null || getConsumer().getRegistries().isEmpty())
                    && (getApplication() == null || getApplication().getRegistries() == null || getApplication().getRegistries().isEmpty())) {
                Map<String, RegistryConfig> registryConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, RegistryConfig.class, false, false);
                if (registryConfigMap != null && registryConfigMap.size() > 0) {
                    List<RegistryConfig> registryConfigs = new ArrayList<RegistryConfig>();
                    for (RegistryConfig config : registryConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault().booleanValue()) {
                            registryConfigs.add(config);
                        }
                    }
                    if (registryConfigs != null && !registryConfigs.isEmpty()) {
                        super.setRegistries(registryConfigs);
                    }
                }
            }
            if (getMonitor() == null
                    && (getConsumer() == null || getConsumer().getMonitor() == null)
                    && (getApplication() == null || getApplication().getMonitor() == null)) {
                Map<String, MonitorConfig> monitorConfigMap = applicationContext == null ? null : BeanFactoryUtils.beansOfTypeIncludingAncestors(applicationContext, MonitorConfig.class, false, false);
                if (monitorConfigMap != null && monitorConfigMap.size() > 0) {
                    MonitorConfig monitorConfig = null;
                    for (MonitorConfig config : monitorConfigMap.values()) {
                        if (config.isDefault() == null || config.isDefault().booleanValue()) {
                            if (monitorConfig != null) {
                                throw new IllegalStateException("Duplicate monitor configs: " + monitorConfig + " and " + config);
                            }
                            monitorConfig = config;
                        }
                    }
                    if (monitorConfig != null) {
                        setMonitor(monitorConfig);
                    }
                }
            }
            Boolean b = isInit();
            if (b == null && getConsumer() != null) {
                b = getConsumer().isInit();
            }
            if (b != null && b.booleanValue()) {
                getObject();
            }
        }
    
        @Override
        public void destroy() {
            // do nothing
        }
    }

afterPropertiesSet初始化，调用getObject方法获取bean

ReferenceConfig

public synchronized T get() {
            if (destroyed) {
                throw new IllegalStateException("Already destroyed!");
            }
            if (ref == null) {
                init();
            }
            return ref;
        }
    
    
        private void init() {
            if (initialized) {
                return;
            }
            initialized = true;
            if (interfaceName == null || interfaceName.length() == 0) {
                throw new IllegalStateException("<dubbo:reference interface=\"\" /> interface not allow null!");
            }
            // get consumer's global configuration
            checkDefault();
            appendProperties(this);
            if (getGeneric() == null && getConsumer() != null) {
                setGeneric(getConsumer().getGeneric());
            }
            if (ProtocolUtils.isGeneric(getGeneric())) {
                interfaceClass = GenericService.class;
            } else {
                try {
                    interfaceClass = Class.forName(interfaceName, true, Thread.currentThread()
                            .getContextClassLoader());
                } catch (ClassNotFoundException e) {
                    throw new IllegalStateException(e.getMessage(), e);
                }
                checkInterfaceAndMethods(interfaceClass, methods);
            }
            String resolve = System.getProperty(interfaceName);
            String resolveFile = null;
            if (resolve == null || resolve.length() == 0) {
                resolveFile = System.getProperty("dubbo.resolve.file");
                if (resolveFile == null || resolveFile.length() == 0) {
                    File userResolveFile = new File(new File(System.getProperty("user.home")), "dubbo-resolve.properties");
                    if (userResolveFile.exists()) {
                        resolveFile = userResolveFile.getAbsolutePath();
                    }
                }
                if (resolveFile != null && resolveFile.length() > 0) {
                    Properties properties = new Properties();
                    FileInputStream fis = null;
                    try {
                        fis = new FileInputStream(new File(resolveFile));
                        properties.load(fis);
                    } catch (IOException e) {
                        throw new IllegalStateException("Unload " + resolveFile + ", cause: " + e.getMessage(), e);
                    } finally {
                        try {
                            if (null != fis) fis.close();
                        } catch (IOException e) {
                            logger.warn(e.getMessage(), e);
                        }
                    }
                    resolve = properties.getProperty(interfaceName);
                }
            }
            if (resolve != null && resolve.length() > 0) {
                url = resolve;
                if (logger.isWarnEnabled()) {
                    if (resolveFile != null && resolveFile.length() > 0) {
                        logger.warn("Using default dubbo resolve file " + resolveFile + " replace " + interfaceName + "" + resolve + " to p2p invoke remote service.");
                    } else {
                        logger.warn("Using -D" + interfaceName + "=" + resolve + " to p2p invoke remote service.");
                    }
                }
            }
            if (consumer != null) {
                if (application == null) {
                    application = consumer.getApplication();
                }
                if (module == null) {
                    module = consumer.getModule();
                }
                if (registries == null) {
                    registries = consumer.getRegistries();
                }
                if (monitor == null) {
                    monitor = consumer.getMonitor();
                }
            }
            if (module != null) {
                if (registries == null) {
                    registries = module.getRegistries();
                }
                if (monitor == null) {
                    monitor = module.getMonitor();
                }
            }
            if (application != null) {
                if (registries == null) {
                    registries = application.getRegistries();
                }
                if (monitor == null) {
                    monitor = application.getMonitor();
                }
            }
            checkApplication();
            checkStubAndMock(interfaceClass);
            Map<String, String> map = new HashMap<String, String>();
            Map<Object, Object> attributes = new HashMap<Object, Object>();
            map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
            map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion());
            map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis()));
            if (ConfigUtils.getPid() > 0) {
                map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid()));
            }
            if (!isGeneric()) {
                String revision = Version.getVersion(interfaceClass, version);
                if (revision != null && revision.length() > 0) {
                    map.put("revision", revision);
                }
    
                String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
                if (methods.length == 0) {
                    logger.warn("NO method found in service interface " + interfaceClass.getName());
                    map.put("methods", Constants.ANY_VALUE);
                } else {
                    map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
                }
            }
            map.put(Constants.INTERFACE_KEY, interfaceName);
            appendParameters(map, application);
            appendParameters(map, module);
            appendParameters(map, consumer, Constants.DEFAULT_KEY);
            appendParameters(map, this);
            String prefix = StringUtils.getServiceKey(map);
            if (methods != null && !methods.isEmpty()) {
                for (MethodConfig method : methods) {
                    appendParameters(map, method, method.getName());
                    String retryKey = method.getName() + ".retry";
                    if (map.containsKey(retryKey)) {
                        String retryValue = map.remove(retryKey);
                        if ("false".equals(retryValue)) {
                            map.put(method.getName() + ".retries", "0");
                        }
                    }
                    appendAttributes(attributes, method, prefix + "." + method.getName());
                    checkAndConvertImplicitConfig(method, map, attributes);
                }
            }
    
            String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
            if (hostToRegistry == null || hostToRegistry.length() == 0) {
                hostToRegistry = NetUtils.getLocalHost();
            } else if (isInvalidLocalHost(hostToRegistry)) {
                throw new IllegalArgumentException("Specified invalid registry ip from property:" + Constants.DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
            }
            map.put(Constants.REGISTER_IP_KEY, hostToRegistry);
    
            //attributes are stored by system context.
            StaticContext.getSystemContext().putAll(attributes);
            ref = createProxy(map);
            ConsumerModel consumerModel = new ConsumerModel(getUniqueServiceName(), this, ref, interfaceClass.getMethods());
            ApplicationModel.initConsumerModel(getUniqueServiceName(), consumerModel);
        }

ReferenceConfig创建代理

@SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
        private T createProxy(Map<String, String> map) {
            URL tmpUrl = new URL("temp", "localhost", 0, map);
            final boolean isJvmRefer;
            if (isInjvm() == null) {
                if (url != null && url.length() > 0) { // if a url is specified, don't do local reference
                    isJvmRefer = false;
                } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
                    // by default, reference local service if there is
                    isJvmRefer = true;
                } else {
                    isJvmRefer = false;
                }
            } else {
                isJvmRefer = isInjvm().booleanValue();
            }
    
            if (isJvmRefer) {
                URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
                invoker = refprotocol.refer(interfaceClass, url);
                if (logger.isInfoEnabled()) {
                    logger.info("Using injvm service " + interfaceClass.getName());
                }
            } else {
                if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
                    String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
                    if (us != null && us.length > 0) {
                        for (String u : us) {
                            URL url = URL.valueOf(u);
                            if (url.getPath() == null || url.getPath().length() == 0) {
                                url = url.setPath(interfaceName);
                            }
                            if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                                urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                            } else {
                                urls.add(ClusterUtils.mergeUrl(url, map));
                            }
                        }
                    }
                } else { // assemble URL from register center's configuration
                    List<URL> us = loadRegistries(false);
                    if (us != null && !us.isEmpty()) {
                        for (URL u : us) {
                            URL monitorUrl = loadMonitor(u);
                            if (monitorUrl != null) {
                                map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                            }
                            urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                        }
                    }
                    if (urls == null || 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) {
                    invoker = refprotocol.refer(interfaceClass, urls.get(0));
                } else {
                    List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
                    URL registryURL = null;
                    for (URL url : urls) {
                        invokers.add(refprotocol.refer(interfaceClass, url));
                        if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                            registryURL = url; // use last registry url
                        }
                    }
                    if (registryURL != null) { // registry url is available
                        // use AvailableCluster only when register's cluster is available
                        URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                        invoker = cluster.join(new StaticDirectory(u, invokers));
                    } else { // not a registry url
                        invoker = cluster.join(new StaticDirectory(invokers));
                    }
                }
            }
    
            Boolean c = check;
            if (c == null && consumer != null) {
                c = consumer.isCheck();
            }
            if (c == null) {
                c = true; // default true
            }
            if (c && !invoker.isAvailable()) {
                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());
            }
            if (logger.isInfoEnabled()) {
                logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
            }
            // create service proxy
            return (T) proxyFactory.getProxy(invoker);
        }

DubboProtocol

public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
            optimizeSerialization(url);
            // create rpc invoker.
            DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
            invokers.add(invoker);
            return invoker;
        }

DubboInvoker

public class DubboInvoker<T> extends AbstractInvoker<T> {
        public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
            super(serviceType, url, new String[]{Constants.INTERFACE_KEY, Constants.GROUP_KEY, Constants.TOKEN_KEY, Constants.TIMEOUT_KEY});
            this.clients = clients;
            // get version.
            this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
            this.invokers = invokers;
        }
    
        @Override
        protected Result doInvoke(final Invocation invocation) throws Throwable {
            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];
            }
            try {
                boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
                boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
                int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
                if (isOneway) {
                    boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
                    currentClient.send(inv, isSent);
                    RpcContext.getContext().setFuture(null);
                    return new RpcResult();
                } else if (isAsync) {
                    ResponseFuture future = currentClient.request(inv, timeout);
                    RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
                    return new RpcResult();
                } else {
                    RpcContext.getContext().setFuture(null);
                    return (Result) currentClient.request(inv, timeout).get();
                }
            } catch (TimeoutException e) {
                throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            } catch (RemotingException e) {
                throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
            }
        }

AbstractInvoker

public Result invoke(Invocation inv) throws RpcException {
            if (destroyed.get()) {
                throw new RpcException("Rpc invoker for service " + this + " on consumer " + NetUtils.getLocalHost()
                        + " use dubbo version " + Version.getVersion()
                        + " is DESTROYED, can not be invoked any more!");
            }
            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);
    
    
            try {
                return doInvoke(invocation);
            } catch (InvocationTargetException e) { // biz exception
                Throwable te = e.getTargetException();
                if (te == null) {
                    return new RpcResult(e);
                } else {
                    if (te instanceof RpcException) {
                        ((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
                    }
                    return new RpcResult(te);
                }
            } catch (RpcException e) {
                if (e.isBiz()) {
                    return new RpcResult(e);
                } else {
                    throw e;
                }
            } catch (Throwable e) {
                return new RpcResult(e);
            }
        }

doInvoke模板方法，子类DubboInvoker实现

Invoker接口，核心是invoke方法

public interface Invoker<T> extends Node {
    
        /**
         * get service interface.
         *
         * @return service interface.
         */
        Class<T> getInterface();
    
        /**
         * invoke.
         *
         * @param invocation
         * @return result
         * @throws RpcException
         */
        Result invoke(Invocation invocation) throws RpcException;
    
    }

然后看createProxy方法的最后面,getProxy  
AbstractProxyFactory

public abstract class AbstractProxyFactory implements ProxyFactory {
    
        public <T> T getProxy(Invoker<T> invoker) throws RpcException {
            Class<?>[] interfaces = null;
            String config = invoker.getUrl().getParameter("interfaces");
            if (config != null && config.length() > 0) {
                String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
                if (types != null && types.length > 0) {
                    interfaces = new Class<?>[types.length + 2];
                    interfaces[0] = invoker.getInterface();
                    interfaces[1] = EchoService.class;
                    for (int i = 0; i < types.length; i++) {
                        interfaces[i + 1] = ReflectUtils.forName(types[i]);
                    }
                }
            }
            if (interfaces == null) {
                interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class};
            }
            return getProxy(invoker, interfaces);
        }
    
        public abstract <T> T getProxy(Invoker<T> invoker, Class<?>[] types);
    
    }

JavassistProxyFactory

public class JavassistProxyFactory extends AbstractProxyFactory {
    
        @SuppressWarnings("unchecked")
        public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
            return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
        }
    
        public <T> Invoker<T> getInvoker(T proxy, Class<T> 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<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);
                }
            };
        }
    
    }