dubbo原理系列1-服务端暴露过程

清疚 2022-06-11 03:54 306阅读 0赞

基础要求：
1 使用dubbo开发过项目
2 了解spring的NamespaceHandler
3 了解一些Netty基本API

Now，进入主题吧。

一般dubbo暴露服务的配置如下

  1. <?xml version="1.0" encoding="UTF-8"?>
  2. <beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:dubbo="http://code.alibabatech.com/schema/dubbo" xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://code.alibabatech.com/schema/dubbo http://code.alibabatech.com/schema/dubbo/dubbo.xsd ">
  3. <dubbo:service interface="**.*ExportService" ref="itemExportService" version="${dubbo.version}" timeout="${dubbo.timeout}" />
  4. </beans>

** 处替换成你自己的服务接口完整路径哦。

dubbo的启动完全依赖spring的容器启动，有两种方式触发

1 通过com.alibaba.dubbo.container.Main这个类入口

  1. public static void main(String[] args) {
  2.         try {
  3.             if (args == null || args.length == 0) {
  4.                 String config = ConfigUtils.getProperty(CONTAINER_KEY, loader.getDefaultExtensionName());
  5.                 args = Constants.COMMA_SPLIT_PATTERN.split(config);
  6.             }
  8.             final List<Container> containers = new ArrayList<Container>();
  9.             for (int i = 0; i < args.length; i ++) {
  10.                 containers.add(loader.getExtension(args[i]));
  11.             }
  12.             logger.info("Use container type(" + Arrays.toString(args) + ") to run dubbo serivce.");
  14.             if ("true".equals(System.getProperty(SHUTDOWN_HOOK_KEY))) {
  15.                 Runtime.getRuntime().addShutdownHook(new Thread() {
  16.                     public void run() {
  17.                         for (Container container : containers) {
  18.                             try {
  19.                                 container.stop();
  20.                                 logger.info("Dubbo " + container.getClass().getSimpleName() + " stopped!");
  21.                             } catch (Throwable t) {
  22.                                 logger.error(t.getMessage(), t);
  23.                             }
  24.                             synchronized (Main.class) {
  25.                                 running = false;
  26.                                 Main.class.notify();
  27.                             }
  28.                         }
  29.                     }
  30.                 });
  31.             }
  33.             for (Container container : containers) {
  34.                 container.start();
  35.                 logger.info("Dubbo " + container.getClass().getSimpleName() + " started!");
  36.             }
  37.             System.out.println(new SimpleDateFormat("[yyyy-MM-dd HH:mm:ss]").format(new Date()) + " Dubbo service server started!");
  38.         } catch (RuntimeException e) {
  39.             e.printStackTrace();
  40.             logger.error(e.getMessage(), e);
  41.             System.exit(1);
  42.         }
  43.         synchronized (Main.class) {
  44.             while (running) {
  45.                 try {
  46.                     Main.class.wait();
  47.                 } catch (Throwable e) {
  48.                 }
  49.             }
  50.         }
  51.     }

这里写图片描述

Main会执行SpringContainer的start方法

  1. public void start() {
  2.         String configPath = ConfigUtils.getProperty(SPRING_CONFIG);
  3.         if (configPath == null || configPath.length() == 0) {
  4.             configPath = DEFAULT_SPRING_CONFIG;
  5.         }
  6.         context = new ClassPathXmlApplicationContext(configPath.split("[,\\s]+"));
  7.         context.start();
  8.     }

可见只是用ClassPathXmlApplicationContext来触发spring容器启动。

故，我们可以推测出另一种方式便是通过WebApplicationContext

2 tomcat触发WebApplicationContext的initWebApplicationContext，从而拉起spring容器

进入重要阶段，容器启动触发dubbo初始化

spring容器启动，触发DubboNamespaceHandler

  1. public void init() {
  2.         registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
  3.         registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
  4.         registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
  5.         registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
  6.         registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
  7.         registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
  8.         registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
  9.         registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
  10.         registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
  11.         registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
  12.     }

注意ServiceBean和AnnotationBean

这两个bean都继承了ServiceConfig，且ServiceBean 实现了ApplicationListener接口,我们知道容器启动时，会触发onApplicationEvent方法

  1. public void onApplicationEvent(ApplicationEvent event) {
  2.         if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
  3.             if (isDelay() && ! isExported() && ! isUnexported()) {
  4.                 if (logger.isInfoEnabled()) {
  5.                     logger.info("The service ready on spring started. service: " + getInterface());
  6.                 }
  7.                 export();
  8.             }
  9.         }
  10.     }

ServiceConfig的export方法

  1. public synchronized void export() {
  2.         if (provider != null) {
  3.             if (export == null) {
  4.                 export = provider.getExport();
  5.             }
  6.             if (delay == null) {
  7.                 delay = provider.getDelay();
  8.             }
  9.         }
  10.         if (export != null && ! export.booleanValue()) {
  11.             return;
  12.         }
  13.         if (delay != null && delay > 0) {
  14.             Thread thread = new Thread(new Runnable() {
  15.                 public void run() {
  16.                     try {
  17.                         Thread.sleep(delay);
  18.                     } catch (Throwable e) {
  19.                     }
  20.                     doExport();
  21.                 }
  22.             });
  23.             thread.setDaemon(true);
  24.             thread.setName("DelayExportServiceThread");
  25.             thread.start();
  26.         } else {
  27.             doExport();
  28.         }
  29.     }
  30. ```又调用了ServiceConfig的doExport方法，doExport调用doExportUrls，doExportUrls调用doExportUrlsFor1Protocol
  31. ，doExportUrlsFor1Protocol有如下代码片段
  36. <div class="se-preview-section-delimiter"></div>
  38. ```java
  39.   if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
  40.                 if (logger.isInfoEnabled()) {
  41.                     logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
  42.                 }
  43.                 if (registryURLs != null && registryURLs.size() > 0
  44.                         && url.getParameter("register", true)) {
  45.                     for (URL registryURL : registryURLs) {
  46.                         url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
  47.                         URL monitorUrl = loadMonitor(registryURL);
  48.                         if (monitorUrl != null) {
  49.                             url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
  50.                         }
  51.                         if (logger.isInfoEnabled()) {
  52.                             logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
  53.                         }
  54.                         Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
  56.                         Exporter<?> exporter = protocol.export(invoker);
  57.                         exporters.add(exporter);
  58.                     }
  59.                 } else {
  60.                     Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
  62.                     Exporter<?> exporter = protocol.export(invoker);
  63.                     exporters.add(exporter);
  64.                 }
  65.             }

重点在 Exporter

  1. if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
  2.                 if (logger.isInfoEnabled()) {
  3.                     logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
  4.                 }
  5.                 if (registryURLs != null && registryURLs.size() > 0
  6.                         && url.getParameter("register", true)) {
  7.                     for (URL registryURL : registryURLs) {
  8.                         url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));
  9.                         URL monitorUrl = loadMonitor(registryURL);
  10.                         if (monitorUrl != null) {
  11.                             url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
  12.                         }
  13.                         if (logger.isInfoEnabled()) {
  14.                             logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
  15.                         }
  16.                         Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
  18.                         Exporter<?> exporter = protocol.export(invoker);
  19.                         exporters.add(exporter);
  20.                     }
  21.                 } else {
  22.                     Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
  24.                     Exporter<?> exporter = protocol.export(invoker);
  25.                     exporters.add(exporter);
  26.                 }
  27.             }

重点在 protocol.export(invoker);

这里写图片描述

默认会进入DubboProtocol

  1. public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  2.         URL url = invoker.getUrl();
  4.         // export service.
  5.         String key = serviceKey(url);
  6.         DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
  7.         exporterMap.put(key, exporter);
  9.         //export an stub service for dispaching event
  10.         Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY,Constants.DEFAULT_STUB_EVENT);
  11.         Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
  12.         if (isStubSupportEvent && !isCallbackservice){
  13.             String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
  14.             if (stubServiceMethods == null || stubServiceMethods.length() == 0 ){
  15.                 if (logger.isWarnEnabled()){
  16.                     logger.warn(new IllegalStateException("consumer [" +url.getParameter(Constants.INTERFACE_KEY) +
  17.                             "], has set stubproxy support event ,but no stub methods founded."));
  18.                 }
  19.             } else {
  20.                 stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
  21.             }
  22.         }
  24.         openServer(url);
  26.         return exporter;
  27.     }

重点在 openServer(url)

  1. private void openServer(URL url) {
  2.         // find server.
  3.         String key = url.getAddress();
  4.         //client 也可以暴露一个只有server可以调用的服务。
  5.         boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true);
  6.         if (isServer) {
  7.             ExchangeServer server = serverMap.get(key);
  8.             if (server == null) {
  9.                 serverMap.put(key, createServer(url));
  10.             } else {
  11.                 //server支持reset,配合override功能使用
  12.                 server.reset(url);
  13.             }
  14.         }
  15.     }

openServer又调用了createServer

  1. private ExchangeServer createServer(URL url) {
  2.         //默认开启server关闭时发送readonly事件
  3.         url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
  4.         //默认开启heartbeat
  5.         url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
  6.         String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
  8.         if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
  9.             throw new RpcException("Unsupported server type: " + str + ", url: " + url);
  11.         url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
  12.         ExchangeServer server;
  13.         try {
  14.             server = Exchangers.bind(url, requestHandler);
  15.         } catch (RemotingException e) {
  16.             throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
  17.         }
  18.         str = url.getParameter(Constants.CLIENT_KEY);
  19.         if (str != null && str.length() > 0) {
  20.             Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
  21.             if (!supportedTypes.contains(str)) {
  22.                 throw new RpcException("Unsupported client type: " + str);
  23.             }
  24.         }
  25.         return server;
  26.     }

createServer又调用了server = Exchangers.bind(url, requestHandler);

Exchangers

  1. public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
  2.         if (url == null) {
  3.             throw new IllegalArgumentException("url == null");
  4.         }
  5.         if (handler == null) {
  6.             throw new IllegalArgumentException("handler == null");
  7.         }
  8.         url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
  9.         return getExchanger(url).bind(url, handler);
  10.     }

Exchangers 调用了HeaderExchanger的bind方法

  1. public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
  2.         return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
  3.     }

重点看Transporters.bind方法

Transporters

  1. public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
  2.         if (url == null) {
  3.             throw new IllegalArgumentException("url == null");
  4.         }
  5.         if (handlers == null || handlers.length == 0) {
  6.             throw new IllegalArgumentException("handlers == null");
  7.         }
  8.         ChannelHandler handler;
  9.         if (handlers.length == 1) {
  10.             handler = handlers[0];
  11.         } else {
  12.             handler = new ChannelHandlerDispatcher(handlers);
  13.         }
  14.         return getTransporter().bind(url, handler);
  15.     }

这里写图片描述

默认会进入NettyTransporter的bind方法

  1. public class NettyTransporter implements Transporter { 
  3.     public static final String NAME = "netty";
  5.     public Server bind(URL url, ChannelHandler listener) throws RemotingException {
  6.         return new NettyServer(url, listener);
  7.     }
  9.     public Client connect(URL url, ChannelHandler listener) throws RemotingException {
  10.         return new NettyClient(url, listener);
  11.     }
  13. }

NettyServer

首先调用了父类AbstractServer构造方法

  1. public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
  2.         super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
  3.     }
  5. public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
  6.         super(url, handler);
  7.         localAddress = getUrl().toInetSocketAddress();
  8.         String host = url.getParameter(Constants.ANYHOST_KEY, false) 
  9.                         || NetUtils.isInvalidLocalHost(getUrl().getHost()) 
  10.                         ? NetUtils.ANYHOST : getUrl().getHost();
  11.         bindAddress = new InetSocketAddress(host, getUrl().getPort());
  12.         this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
  13.         this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
  14.         try {
  15.             doOpen();
  16.             if (logger.isInfoEnabled()) {
  17.                 logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
  18.             }
  19.         } catch (Throwable t) {
  20.             throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() 
  21.                                         + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
  22.         }
  23.         if (handler instanceof WrappedChannelHandler ){
  24.             executor = ((WrappedChannelHandler)handler).getExecutor();
  25.         }
  26.     }

可以看到父类构造方法中调用了doOpen方法后就输出了Start的日志，可见doOpen方法中就是启动了服务端，做好接受调用准备

  1. @Override
  2.     protected void doOpen() throws Throwable {
  3.         NettyHelper.setNettyLoggerFactory();
  4.         ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
  5.         ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
  6.         ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
  7.         bootstrap = new ServerBootstrap(channelFactory);
  9.         final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
  10.         channels = nettyHandler.getChannels();
  11.         // https://issues.jboss.org/browse/NETTY-365
  12.         // https://issues.jboss.org/browse/NETTY-379
  13.         // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
  14.         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
  15.             public ChannelPipeline getPipeline() {
  16.                 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
  17.                 ChannelPipeline pipeline = Channels.pipeline();
  18.                 /*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); }*/
  19.                 pipeline.addLast("decoder", adapter.getDecoder());
  20.                 pipeline.addLast("encoder", adapter.getEncoder());
  21.                 pipeline.addLast("handler", nettyHandler);
  22.                 return pipeline;
  23.             }
  24.         });
  25.         // bind
  26.         channel = bootstrap.bind(getBindAddress());
  27.     }

实际也就是这样，doOpen方法利用Netty开启了服务端的监听，并绑定到InetSocketAddress（getBindAddress()方法返回），这个对象中包含了服务的地址端口。

发表评论

表情：
评论列表 （有 0 条评论，306人围观）

还没有评论，来说两句吧...

相关阅读

    相关 Dubbo服务暴露原理

    服务暴露原理 ![这里写图片描述][70] 配置文件 IOC容器启动，加载配置文件的时候 Dubbo标签处理器，解析每一个标签 封装成对应的组件 ![这

    傷城~傷城~/ 0 赞/ 367 阅读