dubbo原理系列2-reference代理生成过程

逃离我推掉我的手 2022-06-11 05:30 141阅读 0赞

> 这篇文章主要介绍dubbo消费者调用dubbo服务时，服务端代理类（在消费侧实际调用的是dubbo服务的代理类）生成的过程。  
>   
> 开始吧。

入口依然在DubboNamespaceHandler，如果想知道为什么会进入这里，你可以去了解下spring的NamespaceHandler。

public class DubboNamespaceHandler extends NamespaceHandlerSupport { 
    
        static {
            Version.checkDuplicate(DubboNamespaceHandler.class);
        }
    
        public void init() {
            registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
            registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
            registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
            registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
            registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
            registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
            registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
            registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
            registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
            registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
        }
    
    }

重点在这一行

registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));

先解释个疑问，在如下分析的流程代码，以及上一篇梳理  
ServiceBean的代码逻辑，你会在AnnotationBean中也可以看到，不用奇怪，AnnotationBean是另一种配置方式，spring中配置后，会自动解析。

言归正传。

### ReferenceBean的afterPropertiesSet方法 ###

ReferenceBean实现了InitializingBean接口，那么spring容器加载该类的时候，会触发执行afterPropertiesSet执行

public void afterPropertiesSet() throws Exception {
         ...
         Boolean b = isInit();
            if (b == null && getConsumer() != null) {
                b = getConsumer().isInit();
            }
            if (b != null && b.booleanValue()) {
                getObject();
            }
       }

前面好多代码多是做校验与检查，重点看getObject方法

public Object getObject() throws Exception {
            return get();
        }

public synchronized T get() {
            if (destroyed){
                throw new IllegalStateException("Already destroyed!");
            }
            if (ref == null) {
                init();
            }
            return ref;
        }

private void init() {
     ...
        //attributes通过系统context进行存储.
         StaticContext.getSystemContext().putAll(attributes);
         ref = createProxy(map);
    }

核心方法

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) { //指定URL的情况下，不做本地引用
                    isJvmRefer = false;
                } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
                    //默认情况下如果本地有服务暴露，则引用本地服务.
                    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) { // 用户指定URL，指定的URL可能是对点对直连地址，也可能是注册中心URL
                    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 { // 通过注册中心配置拼装URL
                    List<URL> us = loadRegistries(false);
                    if (us != null && us.size() > 0) {
                        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.size() == 0) {
                        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; // 用了最后一个registry url
                        }
                    }
                    if (registryURL != null) { // 有 注册中心协议的URL
                        // 对有注册中心的Cluster 只用 AvailableCluster
                        URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME); 
                        invoker = cluster.join(new StaticDirectory(u, invokers));
                    }  else { // 不是 注册中心的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());
            }
            // 创建服务代理
            return (T) proxyFactory.getProxy(invoker);
        }

这里调用的两个方法要注意  
第一个 refprotocol.refer(…)

![这里写图片描述][SouthEast]

默认是DubboProtocol

### DubboProtocol的refer方法 ###

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

看到这里实际放的是DubboInvoker，返回结果也就是Invoker，这个Invoker是dubbo对服务的抽象代理，那么在客户端调用服务方的接口方法的时候，实际触发的就是Invoker的invoke方法。

*你肯定看到了DubboInvoker的构造函数中第三个参数getClients类型是ExchangeClient集合，也即是dubbo客户端和服务端连接的封装类*

### dubbo客户端与服务端连接的创建 ###

private ExchangeClient[] getClients(URL url){
            //是否共享连接
            boolean service_share_connect = false;
            int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
            //如果connections不配置，则共享连接，否则每服务每连接
            if (connections == 0){
                service_share_connect = true;
                connections = 1;
            }
    
            ExchangeClient[] clients = new ExchangeClient[connections];
            for (int i = 0; i < clients.length; i++) {
                if (service_share_connect){
                    clients[i] = getSharedClient(url);
                } else {
                    clients[i] = initClient(url);
                }
            }
            return clients;
        }

默认是共享连接，会进入getSharedClient

/** *获取共享连接 */
        private ExchangeClient getSharedClient(URL url){
            String key = url.getAddress();
            ReferenceCountExchangeClient client = referenceClientMap.get(key);
            if ( client != null ){
                if ( !client.isClosed()){
                    client.incrementAndGetCount();
                    return client;
                } else {
    // logger.warn(new IllegalStateException("client is closed,but stay in clientmap .client :"+ client));
                    referenceClientMap.remove(key);
                }
            }
            ExchangeClient exchagneclient = initClient(url);
    
            client = new ReferenceCountExchangeClient(exchagneclient, ghostClientMap);
            referenceClientMap.put(key, client);
            ghostClientMap.remove(key);
            return client; 
        }

看看initClient方法

private ExchangeClient initClient(URL url) {
    
            // client type setting.
            String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
    
            String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
            boolean compatible = (version != null && version.startsWith("1.0."));
            url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
            //默认开启heartbeat
            url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
    
            // BIO存在严重性能问题，暂时不允许使用
            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 {
                //设置连接应该是lazy的 
                if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)){
                    client = new LazyConnectExchangeClient(url ,requestHandler);
                } else {
                    client = Exchangers.connect(url ,requestHandler);
                }
            } catch (RemotingException e) {
                throw new RpcException("Fail to create remoting client for service(" + url
                        + "): " + e.getMessage(), e);
            }
            return client;
        }

Exchangers.connect(url ,requestHandler);这一行

public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
            if (url == null) {
                throw new IllegalArgumentException("url == null");
            }
            if (handler == null) {
                throw new IllegalArgumentException("handler == null");
            }
            url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
            return getExchanger(url).connect(url, handler);
        }

public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
            return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
        }

![这里写图片描述][SouthEast 1]

默认进入NettyTransporter的connect方法

public Client connect(URL url, ChannelHandler listener) throws RemotingException {
            return new NettyClient(url, listener);
        }

NettyClient构造方法会依次调用doOpen方法，doConnect方法

### NettyClient ###

@Override
        protected void doOpen() throws Throwable {
            NettyHelper.setNettyLoggerFactory();
            bootstrap = new ClientBootstrap(channelFactory);
            // config
            // @see org.jboss.netty.channel.socket.SocketChannelConfig
            bootstrap.setOption("keepAlive", true);
            bootstrap.setOption("tcpNoDelay", true);
            bootstrap.setOption("connectTimeoutMillis", getTimeout());
            final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
            bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
                public ChannelPipeline getPipeline() {
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
                    ChannelPipeline pipeline = Channels.pipeline();
                    pipeline.addLast("decoder", adapter.getDecoder());
                    pipeline.addLast("encoder", adapter.getEncoder());
                    pipeline.addLast("handler", nettyHandler);
                    return pipeline;
                }
            });
        }

doConnect

protected void doConnect() throws Throwable {
            long start = System.currentTimeMillis();
            ChannelFuture future = bootstrap.connect(getConnectAddress());
            try{
                boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
    
                if (ret && future.isSuccess()) {
                    Channel newChannel = future.getChannel();
                    newChannel.setInterestOps(Channel.OP_READ_WRITE);
                    try {
                        // 关闭旧的连接
                        Channel oldChannel = NettyClient.this.channel; // copy reference
                        if (oldChannel != null) {
                            try {
                                if (logger.isInfoEnabled()) {
                                    logger.info("Close old netty channel " + oldChannel + " on create new netty channel " + newChannel);
                                }
                                oldChannel.close();
                            } finally {
                                NettyChannel.removeChannelIfDisconnected(oldChannel);
                            }
                        }
                    } finally {
                        if (NettyClient.this.isClosed()) {
                            try {
                                if (logger.isInfoEnabled()) {
                                    logger.info("Close new netty channel " + newChannel + ", because the client closed.");
                                }
                                newChannel.close();
                            } finally {
                                NettyClient.this.channel = null;
                                NettyChannel.removeChannelIfDisconnected(newChannel);
                            }
                        } else {
                            NettyClient.this.channel = newChannel;
                        }
                    }
                } else if (future.getCause() != null) {
                    throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                            + getRemoteAddress() + ", error message is:" + future.getCause().getMessage(), future.getCause());
                } else {
                    throw new RemotingException(this, "client(url: " + getUrl() + ") failed to connect to server "
                            + getRemoteAddress() + " client-side timeout "
                            + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start) + "ms) from netty client "
                            + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion());
                }
            }finally{
                if (! isConnected()) {
                    future.cancel();
                }
            }
        }

到此可看到使用netty创建了连接。

继续回到创建代理类ReferenceConfig的createProxy

![这里写图片描述][SouthEast 2]

会进入StubProxyFactoryWrapper的getProxy方法

### StubProxyFactoryWrapper的getProxy方法 ###

@SuppressWarnings({ "unchecked", "rawtypes" })
        public <T> T getProxy(Invoker<T> invoker) throws RpcException {
            T proxy = proxyFactory.getProxy(invoker);
            if (GenericService.class != invoker.getInterface()) {
                String stub = invoker.getUrl().getParameter(Constants.STUB_KEY, invoker.getUrl().getParameter(Constants.LOCAL_KEY));
                if (ConfigUtils.isNotEmpty(stub)) {
                    Class<?> serviceType = invoker.getInterface();
                    if (ConfigUtils.isDefault(stub)) {
                        if (invoker.getUrl().hasParameter(Constants.STUB_KEY)) {
                            stub = serviceType.getName() + "Stub";
                        } else {
                            stub = serviceType.getName() + "Local";
                        }
                    }
                    try {
                        Class<?> stubClass = ReflectUtils.forName(stub);
                        if (! serviceType.isAssignableFrom(stubClass)) {
                            throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + serviceType.getName());
                        }
                        try {
                            Constructor<?> constructor = ReflectUtils.findConstructor(stubClass, serviceType);
                            proxy = (T) constructor.newInstance(new Object[] {proxy});
                            //export stub service
                            URL url = invoker.getUrl();
                            if (url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT)){
                                url = url.addParameter(Constants.STUB_EVENT_METHODS_KEY, StringUtils.join(Wrapper.getWrapper(proxy.getClass()).getDeclaredMethodNames(), ","));
                                url = url.addParameter(Constants.IS_SERVER_KEY, Boolean.FALSE.toString());
                                try{
                                    export(proxy, (Class)invoker.getInterface(), url);
                                }catch (Exception e) {
                                    LOGGER.error("export a stub service error.", e);
                                }
                            }
                        } catch (NoSuchMethodException e) {
                            throw new IllegalStateException("No such constructor \"public " + stubClass.getSimpleName() + "(" + serviceType.getName() + ")\" in stub implemention class " + stubClass.getName(), e);
                        }
                    } catch (Throwable t) {
                        LOGGER.error("Failed to create stub implemention class " + stub + " in consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", cause: " + t.getMessage(), t);
                        // ignore
                    }
                }
            }
            return proxy;
        }

关注下这一行，后面你会有一个地方用到  
url = url.addParameter(Constants.STUB\_EVENT\_METHODS\_KEY, StringUtils.join(Wrapper.getWrapper(proxy.getClass()).getDeclaredMethodNames(), “,”));

调用export方法

private <T> Exporter<T> export(T instance, Class<T> type, URL url) {
            return protocol.export(proxyFactory.getInvoker(instance, type, url));
        }

![这里写图片描述][SouthEast 3]

可见最终进入DubboProtocol的export方法

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
            URL url = invoker.getUrl();
    
            // export service.
            String key = serviceKey(url);
            DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
            exporterMap.put(key, exporter);
    
            //export an stub service for dispaching event
            Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
            Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
            if (isStubSupportEvent && !isCallbackservice){
                String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
                if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
                    if (logger.isWarnEnabled()){
                        logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
                                "], has set stubproxy support event ,but no stub methods founded."));
                    }
                } else {
                    stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
                }
            }
    
            openServer(url);
    
            return exporter;
        }

看看这一行，知道上面为啥要关注那一行了吧！  
Boolean isStubSupportEvent = url.getParameter(Constants.STUB\_EVENT\_KEY,Constants.DEFAULT\_STUB\_EVENT);

所以走的 if (isStubSupportEvent && !isCallbackservice)里的逻辑，再后面逻辑就和之前一篇类似了。

[SouthEast]: /images/20220611/22e9d8ac288d4be4a3fddd01eacb82e8.png
[SouthEast 1]: /images/20220611/25a14d7c1c6141ed85c6dea42da36e1d.png
[SouthEast 2]: /images/20220611/57c057ead384488192629f4a9de1717f.png
[SouthEast 3]: /images/20220611/83c0229badf24740b1661d2c3b4c222b.png