dubbo服务调用源码流程图

2024-03-05 17:56:23   0  举报





这张dubbo服务调用源码流程图描述了dubbo服务调用的主要过程和核心组件。首先，服务提供者通过暴露服务接口，使服务消费者能够发现和调用该服务。服务消费者在调用服务时，会通过代理对象将请求转换为RPC调用，并将请求参数进行序列化。然后，请求将被发送到服务提供者，服务提供者接收到请求后，会进行反序列化，解析请求参数，并调用实际的服务实现类。服务实现类处理请求后，将结果返回给服务提供者，服务提供者将结果序列化并发送回服务消费者。服务消费者接收到响应后，通过代理对象将响应结果反序列化，并返回给调用者。这个过程中，注册中心负责服务注册与发现，监控中心负责监控服务调用和提供者的状态，协议扩展负责处理请求和响应的协议转换。整个流程清晰展示了dubbo服务调用的核心内容和相关组件。

dubbo服务架构图

源码

微服务

java

服务调用

作者其他创作

大纲/内容

result = this.invoker.invoke(invocation);

getClients(url)

ZookeeperRegistry.doRegister(url);

StubProxyFactoryWrapper.getProxy

initServiceMetadata(consumer);

这里源码与服务导出一致

if (routers.isEmpty())

new ClusterFilterInvoker<>(clusterInvoker)

HeaderExchanger.connect

reference.get();

ConsumerContextFilter

代理对象调用

AbstractClusterInvoker<T> last = buildInterceptorInvoker(new ClusterFilterInvoker<>(clusterInvoker));

初始化服务-应用程序映射

prepareInvocation(invocation);

refreshInvoker(urls);

继承

消费者集群过滤链

ZookeeperRegistry.doSubscribe

创建监听器

invoker.invoke(invocation);

result.setExecutor(executor);

refreshServiceDiscoveryInvoker(latch);

invoker.invoke(rpcInvocation).recreate();

NettyTransporter.connect

channel.send(req);

buildClusterInterceptors(doJoin(directory));

for (RegistryProtocolListener listener : listeners)

路径不存在创建/dubbo/org.apache.dubbo.demo.DemoService/providers

netty4

Registry registry = getRegistry(url);

new InvokerInvocationHandler(invoker)

发送请求

动态生成代理类

将 url 变成调用者

HeaderExchangeClient.request

saveProperties(url);

计算优先级

通过netty发送请求

for (FilterChainBuilder builder : builders)

RegistryChildListenerImpl

shouldJvmRefer

buildReferenceCountExchangeClient(url)

MigrationInvoker.invoke()

通过spi机制加载过滤器

DubboProtocol.refer

asyncResult = (AsyncRpcResult) doInvoke(invocation);

MockClusterInvoker.invoke(invocation);

AbstractClusterInvoker.invoke

return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));

FutureFilter

监听器监听节点发生变化

migrationInvoker.migrateToApplicationFirstInvoker(newRule);

AbstractRegistry.notify

MonitorClusterFilter

准备 rpc 调用

FailbackRegistry.register(url);

case APPLICATION_FIRST:

initServiceAppsMapping(referenceParameters);

ListenerRegistryWrapper.subscribe

return resultInvokers;

provider端是serverconsumer端是client

RegistryDirectory.notify

MonitorFilter

refreshInterfaceInvoker(latch);

RouterSnapshotFilter

路径与监听器绑定

listener.notify(categoryList);

force:direct mock

currentClient.request

DefaultFilterChainBuilder.buildClusterInvokerChain

循环结束唤醒监听

filterInvoker.invoke(invocation);

AsyncRpcResult asyncResult = doInvokeAndReturn(invocation);

构建处理链

registry.register(url);

true

JavassistProxyFactory.getProxy

directory.subscribe(toSubscribeUrl(urlToRegistry));

通过spi机制根据url获取不同的注册中心

createInvokerForRemote();

createInvokerForLocal(referenceParameters);

doInvokeAndReturn(invocation);

List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);

DynamicDirectory.subscribe

createProxy(referenceParameters);

RegistryDirectory.subscribe

Proxy.getProxy

List<Invoker<T>> invokers = list(invocation);

migrationRuleHandler.doMigrate(rule);

for (String path : toCategoriesPath(url))保存url对应的监听器

默认使用随机负载均衡器RandomLoadBalance

AbstractCluster.invoke(invocation);

通过spi机制获取到

初始化服务元数据

init();

notify()

每次收到监听节点变化都会更新本地缓存

initClient(url);

this.currentAvailableInvoker = invoker;

getScopeModel().getDeployer().start();

目标实例集群

doSelect方法

FailbackRegistry.subscribe

currentAvailableInvoker.invoke(invocation);

repository.registerConsumer(consumerModel);

FailoverCluster

MigrationRuleListener.onRefer

doRegister(url);

AbstractInvoker.invoke(invocation);

doSaveProperties(version);

初始化负载均衡器

这里还有一步会创建routerChain

注册consumer

refreshOverrideAndInvoker(providerURLs);

调用

AbstractRegistry.subscribe

创建新连接

ListenerRegistryWrapper.register

InterfaceCompatibleRegistryProtocol.getMigrationInvoker

calcPreferredInvoker(newRule);

保存本地文件缓存

RegistryProtocol.refer()

create rpc invoker.

InvokerInvocationHandler.invoke

String message = service.sayHello(\"dubbo\");

调用 rpc 调用并返回异步结果

registry.register(directory.getRegisteredConsumerUrl());

return asyncResult;

MigrationInvoker.migrateToApplicationFirstInvoker(newRule);

构建消费者集群过滤链

ReferenceConfig<DemoService> reference = new ReferenceConfig<>(); reference.setApplication(new ApplicationConfig(\"dubbo-demo-api-consumer\")); reference.setRegistry(new RegistryConfig(\"zookeeper://127.0.0.1:2181\")); reference.setMetadataReportConfig(new MetadataReportConfig(\"zookeeper://127.0.0.1:2181\")); reference.setInterface(DemoService.class); DemoService service = reference.get(); String message = service.sayHello(\"dubbo\");

AbstractCluster.join

HeaderExchangeChannel.request

AbstractProxyFactory.getProxy

调用get获取结果

Transporters.connect

将请求内容封装成Invocation

等待结果返回

DubboInvoker.doInvoke

aggregateUrlFromRegistry(referenceParameters);

calculateInvokeTimes(methodName);计算调用次数

FailoverClusterInvoker.doInvoke

false