您好,登錄后才能下訂單哦!
這篇文章主要介紹“Dubbo Consumer引用服務(wù)的方法是什么”的相關(guān)知識,小編通過實(shí)際案例向大家展示操作過程,操作方法簡單快捷,實(shí)用性強(qiáng),希望這篇“Dubbo Consumer引用服務(wù)的方法是什么”文章能幫助大家解決問題。
<?xml version="1.0" encoding="UTF-8"?> <beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://dubbo.apache.org/schema/dubbo" xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd"> <context:property-placeholder/> <dubbo:application name="serialization-java-consumer"> <dubbo:parameter key="qos.enable" value="true" /> <dubbo:parameter key="qos.accept.foreign.ip" value="false" /> <dubbo:parameter key="qos.port" value="33333" /> </dubbo:application> <dubbo:registry address="zookeeper://${zookeeper.address:127.0.0.1}:2181"/> <dubbo:reference id="demoService" check="true" interface="org.apache.dubbo.samples.serialization.api.DemoService"/> </beans>
在之前的章節(jié)中已經(jīng)知道,Dubbo基于Spring自定義標(biāo)簽規(guī)范實(shí)現(xiàn)了自定義標(biāo)簽,通過自定義標(biāo)簽完成了bean的加載,并且通過實(shí)現(xiàn)監(jiān)聽Spring容器刷新完畢事件啟動dubbo客戶端。啟動客戶端伴隨著服務(wù)發(fā)布和服務(wù)的訂閱。
public class DubboConsumer { public static void main(String[] args) { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring/dubbo-demo-consumer.xml"); context.start(); DemoService demoService = context.getBean("demoService", DemoService.class); String hello = demoService.sayHello("world"); System.out.println(hello); } }
dubbo通過<dubbo:reference
標(biāo)簽引用服務(wù),之后在程序中通過Spring的Context依賴查找(getBean)的方式獲取引用的服務(wù)的代理實(shí)例。<dubbo:reference
加載的Bean是ReferenceBean,它實(shí)現(xiàn)了FactoryBean接口,getBean時會調(diào)用ReferenceBean的getObject()方法,這是獲取引用的入口。getBean方法會判斷Reference對象是否是空的,如果是空的,調(diào)用init方法。代碼如下:
@Override public Object getObject() { return get(); }
ReferenceBean
繼承了ReferenceConfig
,當(dāng)調(diào)用ReferenceBean的getObject()方法會調(diào)用ReferenceBean
的get()方法。
public synchronized T get() { if (destroyed) { throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!"); } // 代理引用如果是空的,調(diào)用init if (ref == null) { init(); } return ref; } public synchronized void init() { if (initialized) { return; } if (bootstrap == null) { bootstrap = DubboBootstrap.getInstance(); bootstrap.init(); } // 1. 檢查配置ConsumerConfig,有的話檢查配置,沒有就新建一個ConsumerConfig // 2. 反射創(chuàng)建調(diào)用的API // 3. 初始化ServiceMetadata // 4. 注冊Consumer // 5. 檢查ReferenceConfig,RegistryConfig,ConsumerConfig checkAndUpdateSubConfigs(); checkStubAndLocal(interfaceClass); // 檢查引用的接口是否mock ConfigValidationUtils.checkMock(interfaceClass, this); // consumer的信息 Map<String, String> map = new HashMap<String, String>(); map.put(SIDE_KEY, CONSUMER_SIDE); // 添加運(yùn)行時參數(shù)到map,包括:dubbo,release,timestamp,pid ReferenceConfigBase.appendRuntimeParameters(map); // 是不是泛化,不是的話進(jìn)入條件 if (!ProtocolUtils.isGeneric(generic)) { String revision = Version.getVersion(interfaceClass, version); if (revision != null && revision.length() > 0) { map.put(REVISION_KEY, revision); } // 獲取方法,生成包裝類,使用javassist,將生成的類放到WRAPPER_MAP中,key是org.apache.dubbo.samples.serialization.api.DemoService類對象,value是包裝類的實(shí)例 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 { // method放到map,這里method是sayHello map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR)); } } // interface org.apache.dubbo.samples.serialization.api.DemoService map.put(INTERFACE_KEY, interfaceName); // 追加其他參數(shù)到map中 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); // 所有數(shù)據(jù)在存到serviceMetadata的attachments中 serviceMetadata.getAttachments().putAll(map); // 創(chuàng)建Service代理 ref = createProxy(map); // 設(shè)置ServiceMetadata的Service代理引用 serviceMetadata.setTarget(ref); serviceMetadata.addAttribute(PROXY_CLASS_REF, ref); ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey()); consumerModel.setProxyObject(ref); consumerModel.init(attributes); // 標(biāo)記初始化完畢 initialized = true; // dispatch a ReferenceConfigInitializedEvent since 2.7.4 dispatch(new ReferenceConfigInitializedEvent(this, invoker)); }
@SuppressWarnings({"unchecked", "rawtypes", "deprecation"}) private T createProxy(Map<String, String> map) { // 是否是InJvm,協(xié)議是InJvm if (shouldJvmRefer(map)) { 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(); 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 // 如果protocols不是injvm檢查注冊中心 if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) { checkRegistry(); // url列表,將zookeeper://改成registry:// List<URL> us = ConfigValidationUtils.loadRegistries(this, false); if (CollectionUtils.isNotEmpty(us)) { for (URL u : us) { // 監(jiān)控URL URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u); // 如果有監(jiān)控配置放到map中 if (monitorUrl != null) { map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString())); } // 根據(jù)map中的信息生成url,這里生成的結(jié)果是 /* registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=serialization-java-consumer&dubbo=2.0.2&pid=66793&qos.accept.foreign.ip=false&qos.enable=true&qos.port=33333&refer=application%3Dserialization-java-consumer%26check%3Dtrue%26dubbo%3D2.0.2%26init%3Dfalse%26interface%3Dorg.apache.dubbo.samples.serialization.api.DemoService%26methods%3DsayHello%26pid%3D66793%26qos.accept.foreign.ip%3Dfalse%26qos.enable%3Dtrue%26qos.port%3D33333%26register.ip%3D192.168.58.45%26release%3D2.7.7%26side%3Dconsumer%26sticky%3Dfalse%26timestamp%3D1636532992568®istry=zookeeper&release=2.7.7×tamp=1636533091883 */ 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."); } } } // 如果引用的URL就1個直接通過refer引用服務(wù),這里和export相似,通過調(diào)用鏈實(shí)現(xiàn)的 // - ProtocolListenerWrapper // - - ProtocolFilterWrapper // - - - RegistryProtocol if (urls.size() == 1) { // 通過RegistryProtocol#refer引用服務(wù) invoker = REF_PROTOCOL.refer(interfaceClass, urls.get(0)); } else { // 如果是引用的服務(wù)多個,循環(huán)處理 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 default // 集群處理 URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME); // The invoker wrap relation would be like: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker // 加入到集群中,這里包含集中集群處理模式,分別是: invoker = CLUSTER.join(new StaticDirectory(u, invokers)); } else { // not a registry url, must be direct invoke. invoker = CLUSTER.join(new StaticDirectory(invokers)); } } } // invoer不可用處理 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()); } if (logger.isInfoEnabled()) { logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl()); } /** * @since 2.7.0 * ServiceData存儲,SPI機(jī)制,支持內(nèi)存和遠(yuǎn)程兩種方式 */ 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 proxy return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic)); }
創(chuàng)建代理服務(wù)會創(chuàng)建Invoker,在引用服務(wù)過程中,會判斷協(xié)議是否為injvm,會根據(jù)協(xié)議做不同的處理,不是injvm協(xié)議會根據(jù)構(gòu)造的配置信息(map)生成url并將url協(xié)議有zookeeper://改成registry://,然后通過Protocol
接口refer
方法引用服務(wù),與發(fā)布服務(wù)相似,引用服務(wù)的過程也會包裝方法的調(diào)用鏈,如下:
- ProtocolListenerWrapper
- - ProtocolFilterWrapper
- - - RegistryProtocol
在refer的過程中會對一個服務(wù)端的引用和一個服務(wù)多個服務(wù)端的服務(wù)進(jìn)行區(qū)分處理,對于有多個服務(wù)端的服務(wù)會進(jìn)行集群處理(cluster),會講invoker列表加入到集群中,在調(diào)用過程中會根據(jù)集群策略來選擇不同的策略進(jìn)行調(diào)用,集群策略實(shí)現(xiàn)也實(shí)現(xiàn)了SPI機(jī)制,
@Override @SuppressWarnings("unchecked") public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException { // 注冊中心url 將 registry:// 轉(zhuǎn)成 zookeeper:// url = getRegistryUrl(url); // 獲取注冊中心 Registry registry = registryFactory.getRegistry(url); if (RegistryService.class.equals(type)) { return proxyFactory.getInvoker((T) registry, type, url); } // 從url中解析出引用服務(wù)信息 Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY)); // 獲取分組 String group = qs.get(GROUP_KEY); // 如果設(shè)置分組了,那么使用MergeableCluster策略 if (group != null && group.length() > 0) { if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) { return doRefer(getMergeableCluster(), registry, type, url); } } return doRefer(cluster, registry, type, url); }
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) { // 獲取集群目錄Directory,代表Invoker的集合 RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url); directory.setRegistry(registry); directory.setProtocol(protocol); // all attributes of REFER_KEY Map<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); if (directory.isShouldRegister()) { directory.setRegisteredConsumerUrl(subscribeUrl); // 注冊消費(fèi)者url registry.register(directory.getRegisteredConsumerUrl()); } directory.buildRouterChain(subscribeUrl); // 向ZK發(fā)起訂閱服務(wù),并設(shè)置監(jiān)聽,這里比較復(fù)雜,最終會在ZookeeperRegistry#doSubscribe做服務(wù)訂閱 directory.subscribe(toSubscribeUrl(subscribeUrl)); // 加入集群 Invoker<T> invoker = cluster.join(directory); List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url); if (CollectionUtils.isEmpty(listeners)) { return invoker; } RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker, subscribeUrl); for (RegistryProtocolListener listener : listeners) { listener.onRefer(this, registryInvokerWrapper); } return registryInvokerWrapper; }
RegistryProtocol#doRefer
方法,directory.subscribe會按照下面的調(diào)用鏈進(jìn)行處理,最后調(diào)用ZookeeperRegistry#doSubscribe
方法向zk注冊數(shù)據(jù)訂閱接口,并設(shè)置監(jiān)聽。
- RegistryDirectory
- - ListenerRegistryWrapper
- - - FailbackRegistry
- - - - ZookeeperRegistry
@Override public void doSubscribe(final URL url, final NotifyListener listener) { try { if (ANY_VALUE.equals(url.getServiceInterface())) { String root = toRootPath(); // 初始化監(jiān)聽 ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> { for (String child : currentChilds) { child = URL.decode(child); if (!anyServices.contains(child)) { anyServices.add(child); subscribe(url.setPath(child).addParameters(INTERFACE_KEY, child, Constants.CHECK_KEY, String.valueOf(false)), k); } } }); // 創(chuàng)建zk節(jié)點(diǎn) zkClient.create(root, false); List<String> services = zkClient.addChildListener(root, zkListener); if (CollectionUtils.isNotEmpty(services)) { for (String service : services) { service = URL.decode(service); anyServices.add(service); subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service, Constants.CHECK_KEY, String.valueOf(false)), listener); } } } else { List<URL> urls = new ArrayList<>(); for (String path : toCategoriesPath(url)) { ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>()); ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, k, toUrlsWithEmpty(url, parentPath, currentChilds))); zkClient.create(path, false); List<String> children = zkClient.addChildListener(path, zkListener); if (children != null) { urls.addAll(toUrlsWithEmpty(url, path, children)); } } // notify(url, listener, urls); } } catch (Throwable e) { throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); } }
refer服務(wù)創(chuàng)建Invoker時會調(diào)用該方法,該方法會通過getClients
創(chuàng)建網(wǎng)絡(luò)客戶端,創(chuàng)建客戶端是會判斷客戶端是否為共享鏈接,根據(jù)connections
創(chuàng)建客戶端ExchangeClient
,然后通過initClient
初始化客戶端,初始化過程中會判斷是否為延遲的客戶端LazyConnectExchangeClient
,不是延遲客戶端,就會通過connect
連接服務(wù)提供者,與服務(wù)提供者連接,具體建立連接流程不在這里說明,會在網(wǎng)絡(luò)通信介紹。
@Override public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException { optimizeSerialization(url); // 創(chuàng)建RPC Invoker,通過getClients(url)創(chuàng)建網(wǎng)絡(luò)客戶端 DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers); // 將invoker添加到invokers中 invokers.add(invoker); return invoker; } // 創(chuàng)建客戶端數(shù)組 private ExchangeClient[] getClients(URL url) { // whether to share connection // 是否共享鏈接 boolean useShareConnect = false; int connections = url.getParameter(CONNECTIONS_KEY, 0); List<ReferenceCountExchangeClient> shareClients = null; // if not configured, connection is shared, otherwise, one connection for one service if (connections == 0) { useShareConnect = true; /* * The xml configuration should have a higher priority than properties. * xml配置優(yōu)先級高于properties配置 */ String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null); connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY, DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr); // 共享鏈接客戶端 shareClients = getSharedClient(url, connections); } // 創(chuàng)建ExchangeClient ExchangeClient[] clients = new ExchangeClient[connections]; for (int i = 0; i < clients.length; i++) { if (useShareConnect) { // 從共享客戶端獲取 clients[i] = shareClients.get(i); } else { // 初始化客戶端 clients[i] = initClient(url); } } return clients; } // 初始化客戶端 private ExchangeClient initClient(URL url) { // client type setting. String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT)); url = url.addParameter(CODEC_KEY, DubboCodec.NAME); // enable heartbeat by default url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT)); // BIO is not allowed since it has severe performance issue. if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException("Unsupported client type: " + str + "," + " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " ")); } ExchangeClient client; try { // connection should be lazy if (url.getParameter(LAZY_CONNECT_KEY, false)) { // 延遲加載客戶端 client = new LazyConnectExchangeClient(url, requestHandler); } else { // 通過NettyTransporter connect創(chuàng)建客戶端 與provider建立連接 client = Exchangers.connect(url, requestHandler); } } catch (RemotingException e) { throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e); } return client; }
關(guān)于“Dubbo Consumer引用服務(wù)的方法是什么”的內(nèi)容就介紹到這里了,感謝大家的閱讀。如果想了解更多行業(yè)相關(guān)的知識,可以關(guān)注億速云行業(yè)資訊頻道,小編每天都會為大家更新不同的知識點(diǎn)。
免責(zé)聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點(diǎn)不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進(jìn)行舉報,并提供相關(guān)證據(jù),一經(jīng)查實(shí),將立刻刪除涉嫌侵權(quán)內(nèi)容。