调试dubbo-2.服务导出

分手后的思念是犯贱 2022-11-29 13:22 53阅读 0赞

1.  服务导出前组装URL
    
    导出方法export()经过一些参数校验，最终调用doExportUrls()，如下
    
        private void doExportUrls() { 
                List<URL> registryURLs = loadRegistries(true);
                for (ProtocolConfig protocolConfig : protocols) { 
                    doExportUrlsFor1Protocol(protocolConfig, registryURLs);
                }
            }
    
    加载注册中心配置是在com.alibaba.dubbo.config.AbstractInterfaceConfig\#loadRegistries方法中做的，[上篇文章第三节][Link 1]中已经解析并加载好了<dubbo:registry address=“multicast://224.5.6.7:1234”/>标签对应的RegistryConfig类，从中解析出注册中心地址，再补充一些 服务名application, 应用进程pid, dubbo版本，动态控制服务的qos端口等信息，将这些信息组装成URL返回，也就是这个方法的返回结果List，如果配置了多个注册中心，都会创建对应的URL对象加入到列表中。url的形式大致如下：
    
    registry://224.5.6.7:1234/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&pid=26292&qos.port=22222&registry=multicast&timestamp=1599101771418
    
    protocols与registries类似，通过<dubbo:protocol >标签解析出来的，存在protocols中，如果配置了多种协议，遍历每种协议，导出多协议版本的服务。
    
    doExportUrlsFor1Protocol方法中又是获取很多服务导出必要的参数，比如服务提供方ip地址地址bind.ip，导出方法名，dubbo端口bind.port, 进程pid等，组装成URL，如下
    
    dubbo://10.10.10.10:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.33.69.77&bind.port=20880&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=15916&qos.port=22222&side=provider&timestamp=1599101393956
    
    导出本地服务还是远程服务，根据 url 中的 scope 参数决定，不过无论哪种形式，都要先创建 Invoker, 比如导出本地的代码:
    
        Exporter<?> exporter = protocol.export(
                proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
    
    导出远程代码：
    
        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
        
        Exporter<?> exporter = protocol.export(wrapperInvoker);
    
    都有proxyFactory.getInvoker 创建Invoker。
2.  导出服务时创建Invoker
    
    > Invoker 是实体域，它是 Dubbo 的核心模型，其它模型都向它靠扰，或转换成它，它代表一个可执行体，可向它发起 invoke 调用，它有可能是一个本地的实现，也可能是一个远程的实现，也可能一个集群实现。 ——dubbo文档
    
    proxyFactory 是SPI加载的类，其中默认值指定了javassist
    
        @SPI("javassist")
        public interface ProxyFactory { 
        	//。。。。。。
        }
    
    dubbo自适应给ProxyFactory生成的代理适配类如下：
    
        public class ProxyFactory$Adaptive implements com.alibaba.dubbo.rpc.ProxyFactory { 
        public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.RpcException { 
            if (arg0 == null) 
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
            if (arg0.getUrl() == null) 
                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");
            com.alibaba.dubbo.common.URL url = arg0.getUrl();
            String extName = url.getParameter("proxy", "javassist");
            if(extName == null) 
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
            com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
            return extension.getProxy(arg0);
        }
    
    url.getParameter(“proxy”, “javassist”); 如果url中没有指明用什么类型的代理工厂，那就默认使用javassist指定的，javassist=com.alibaba.dubbo.rpc.proxy.javassist.JavassistProxyFactory，它的getInvoker方法如下：proxy是我们暴露接口的实现类，type是暴露的接口，url就是上面各种参数组装好的url。
    
        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);
                }
            };
        }
    
    Invoker执行业务是就是调用doInvoke方法的，并且这个方法直接交给wrapper.invokeMethod方法执行。wrapper看名字就知道这是个包装类，包装服务方接口的实现类。在getWrapper方法内最终调用com.alibaba.dubbo.common.bytecode.Wrapper\#makeWrapper方法，拼接包装类字符串，通过javassist编译成最终的包装类，生成的invokeMethod大致如下，包括了方法名、参数长度的校验。
    
        public Object invokeMethod(Object o, String n, Class[] p, Object[] v) throws java.lang.reflect.InvocationTargetException{  									        	com.alibaba.dubbo.demo.provider.DemoServiceImpl w; 
            try{  w = ((com.alibaba.dubbo.demo.provider.DemoServiceImpl)$1); }
            catch(Throwable e){  throw new IllegalArgumentException(e); } 
            try{  if( "sayHello".equals( $2 )  &&  $3.length == 1 ) {   
                	return ($w)w.sayHello((java.lang.String)$4[0]); 
            	} 
            } catch(Throwable e) {   
                throw new java.lang.reflect.InvocationTargetException(e);  
            } 
            throw new com.alibaba.dubbo.common.bytecode.NoSuchMethodException("Not found method \""+$2+"\" in class com.alibaba.dubbo.demo.provider.DemoServiceImpl."); 
        }
    
    Invoker创建好之后，通过 protocol.export方法导出服务。
    
    Exporter<?> exporter = protocol.export(  
    proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
3.  把Invoker导出为Exporter
    
    3.1 导出本地时，修改了组装好的url, 代码如下
    
        private void exportLocal(URL url) { 
            if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) { 
                URL local = URL.valueOf(url.toFullString())
                    .setProtocol(Constants.LOCAL_PROTOCOL)
                    .setHost(LOCALHOST)
                    .setPort(0);
                Exporter<?> exporter = protocol.export(
                    proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
                exporters.add(exporter);
                logger.info("Export dubbo service " + interfaceClass.getName() + " to local registry");
            }
        }
    
    修改后的url形式如下：
    
    injvm://127.0.0.1/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.33.69.77&bind.port=20880&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=16264&qos.port=22222&side=provider&timestamp=1599099847351
    
    protocol也是SPI自适应扩展的，根据url中的协议参数，获取到injvm=com.alibaba.dubbo.rpc.protocol.injvm.InjvmProtocol，它的export方法如下, 只创建一个InjvmExporter返回即可。
    
        public class InjvmProtocol extends AbstractProtocol implements Protocol { 
        	// .....
            public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
                return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
            }
        	// .....
        }
    
    3.2 导出远程时
    
        public class ServiceConfig<T> extends AbstractServiceConfig { 
            // ......
            private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { 
        		// ......
                    if (registryURLs != null && registryURLs.size() > 0) { 
                        for (URL registryURL : registryURLs) { 
                            url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
                            URL monitorUrl = loadMonitor(registryURL);
                            if (monitorUrl != null) { 
                                url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
                            }
                            if (logger.isInfoEnabled()) { 
                                logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                            }
                            Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                            DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
        
                            Exporter<?> exporter = protocol.export(wrapperInvoker);
                            exporters.add(exporter);
                        }
                    } 
              // ...... 
            }
        }
    
    遍历registryURLs，如果有多注册中心逐个注册中心地址导出。此时的registryURL是
    
    registry://224.5.6.7:1234/com.alibaba.dubbo.registry.RegistryService?application=demo-provider&dubbo=2.0.0&export=dubbo%3A%2F%2F10.33.69.77%3A20880%2Fcom.alibaba.dubbo.demo.DemoService%3Fanyhost%3Dtrue%26application%3Ddemo-provider%26bind.ip%3D10.33.69.77%26bind.port%3D20880%26dubbo%3D2.0.0%26generic%3Dfalse%26interface%3Dcom.alibaba.dubbo.demo.DemoService%26methods%3DsayHello%26pid%3D6552%26qos.port%3D22222%26side%3Dprovider%26timestamp%3D1599112738462&pid=6552&qos.port=22222&registry=multicast&timestamp=1599112738436
    
    自适应加载registry=com.alibaba.dubbo.registry.integration.RegistryProtocol, 它的export方法如下
    
        public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { 
            //export invoker
            final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        	// ......省略，这部分为服务注册，后文再看
        }
    
    doLocalExport(originInvoker)中，取url是origininvoker的parameters中的url，origininvoker.getUrl().getParameterAndDecoded(Constants.EXPORT\_KEY)，此时的url为
    
    dubbo://10.33.69.77:20880/com.alibaba.dubbo.demo.DemoService?anyhost=true&application=demo-provider&bind.ip=10.33.69.77&bind.port=20880&dubbo=2.0.0&generic=false&interface=com.alibaba.dubbo.demo.DemoService&methods=sayHello&pid=16264&qos.port=22222&side=provider&timestamp=1599099847351
    
    自适应加载dubbo=com.alibaba.dubbo.rpc.protocol.dubbo.DubboProtocol，它的export方法中
    
        public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
            URL url = invoker.getUrl();
            // ......
            DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
        	// ......
            openServer(url);
            return exporter;
        }
    
    创建了一个DubboExporter，然后openServer启动服务器。
    
        private void openServer(URL url) { 
            // ......省略
                    serverMap.put(key, createServer(url));
        	// ......省略
        }
4.  导出远程过程中需要openServer启动服务
    
    openServer中createServer创建服务
    
        private ExchangeServer createServer(URL url) { 
            // ......省略
            ExchangeServer server;
            try { 
                server = Exchangers.bind(url, requestHandler);
            } 
            // ......省略
            return server;
        }

再看Exchangers.bind

public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { 
           // ......省略
           url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
           return getExchanger(url).bind(url, handler);
       }
       
       public static Exchanger getExchanger(URL url) { 
           String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
           return getExchanger(type);
       }
       
       public static Exchanger getExchanger(String type) { 
           return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
       }

通过SPI加载Exchanger，type是上面指定的Constants.DEFAULT\_EXCHANGER = “header”, 于是进入HeaderExchanger的bind方法

public class HeaderExchanger implements Exchanger { 
           // ......省略
           public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { 
               return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
           }
       }

Transporters也是通过SPI加载的，@SPI注解默认指定的是netty, 于是加载NettyTransporter类netty=com.alibaba.dubbo.remoting.transport.netty.NettyTransporter

@SPI("netty")
       public interface Transporter { 
       }

NettyTransporter.bind方法中，创建了一个NettyServer

public class NettyTransporter implements Transporter { 
       	// ......省略
           public Server bind(URL url, ChannelHandler listener) throws RemotingException { 
               return new NettyServer(url, listener);
           }
       }

NettyServer继承自AbstractServer, 构造方法调用的父类构造方法, 先来看看构造方法：

public class NettyServer extends AbstractServer implements Server { 
           public NettyServer(URL url, ChannelHandler handler) throws RemotingException { 
               super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
           }
           // ......省略
       }

public abstract class AbstractServer extends AbstractEndpoint implements Server { 
       	// ......省略
           public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { 
               // ......省略
               String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
               int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
               // ......省略
               try { 
                   doOpen();
               } 
               // ......省略
       }

构造方法中设置编码器，超时时间等参数, 另外从Url中取值，比如ip，端口等。之后再doOpen方法中打开netty服务连接。代码如下，其中，创建netty的 boss 和 worker 线程池，创建ServerBootstrap，绑定得到通道，至此，服务导出结束。

public class NettyServer extends AbstractServer implements Server { 
       
           @Override
           protected void doOpen() throws Throwable { 
               NettyHelper.setNettyLoggerFactory();
               ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
               ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
               ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
               bootstrap = new ServerBootstrap(channelFactory);
       
               bootstrap.setOption("child.tcpNoDelay", true);
               bootstrap.setPipelineFactory(new ChannelPipelineFactory() { 
                   // ......省略
               });
               // bind
               channel = bootstrap.bind(getBindAddress());
           }
       }

1.  导出结束后注册服务
    
        public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { 
            // ......省略
            // 在ProviderConsumerRegTable的属性ConcurrentHashMap<String, Set<ProviderInvokerWrapper>> providerInvokers中保存一下Invoker
            ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registedProviderUrl);
        	// ......省略
            if (register) { 
                register(registryUrl, registedProviderUrl);
                ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);
            }
        	// ......省略
            return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registedProviderUrl);
        }
    
    以zookeeper为例，注册服务的主体流就是：
    
     *  创建和zookeeper服务的连接客户端
     *  通过这个客户端在zookeeper的某个路径下创建文件。所谓的服务注册，本质上是将服务配置数据写入到 Zookeeper 的某个路径的节点下

[Link 1]: https://blog.csdn.net/u013041642/article/details/109103551