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

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

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

Now，进入主题吧。

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

<?xml version="1.0" encoding="UTF-8"?>
    <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 ">
    <dubbo:service interface="**.*ExportService" ref="itemExportService" version="${dubbo.version}" timeout="${dubbo.timeout}" />
    </beans>

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

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

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

public static void main(String[] args) {
            try {
                if (args == null || args.length == 0) {
                    String config = ConfigUtils.getProperty(CONTAINER_KEY, loader.getDefaultExtensionName());
                    args = Constants.COMMA_SPLIT_PATTERN.split(config);
                }
    
                final List<Container> containers = new ArrayList<Container>();
                for (int i = 0; i < args.length; i ++) {
                    containers.add(loader.getExtension(args[i]));
                }
                logger.info("Use container type(" + Arrays.toString(args) + ") to run dubbo serivce.");
    
                if ("true".equals(System.getProperty(SHUTDOWN_HOOK_KEY))) {
                    Runtime.getRuntime().addShutdownHook(new Thread() {
                        public void run() {
                            for (Container container : containers) {
                                try {
                                    container.stop();
                                    logger.info("Dubbo " + container.getClass().getSimpleName() + " stopped!");
                                } catch (Throwable t) {
                                    logger.error(t.getMessage(), t);
                                }
                                synchronized (Main.class) {
                                    running = false;
                                    Main.class.notify();
                                }
                            }
                        }
                    });
                }
    
                for (Container container : containers) {
                    container.start();
                    logger.info("Dubbo " + container.getClass().getSimpleName() + " started!");
                }
                System.out.println(new SimpleDateFormat("[yyyy-MM-dd HH:mm:ss]").format(new Date()) + " Dubbo service server started!");
            } catch (RuntimeException e) {
                e.printStackTrace();
                logger.error(e.getMessage(), e);
                System.exit(1);
            }
            synchronized (Main.class) {
                while (running) {
                    try {
                        Main.class.wait();
                    } catch (Throwable e) {
                    }
                }
            }
        }

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

Main会执行SpringContainer的start方法

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

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

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

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

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

### spring容器启动，触发DubboNamespaceHandler ###

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));
        }

注意ServiceBean和AnnotationBean

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

public void onApplicationEvent(ApplicationEvent event) {
            if (ContextRefreshedEvent.class.getName().equals(event.getClass().getName())) {
                if (isDelay() && ! isExported() && ! isUnexported()) {
                    if (logger.isInfoEnabled()) {
                        logger.info("The service ready on spring started. service: " + getInterface());
                    }
                    export();
                }
            }
        }

### ServiceConfig的export方法 ###

public synchronized void export() {
            if (provider != null) {
                if (export == null) {
                    export = provider.getExport();
                }
                if (delay == null) {
                    delay = provider.getDelay();
                }
            }
            if (export != null && ! export.booleanValue()) {
                return;
            }
            if (delay != null && delay > 0) {
                Thread thread = new Thread(new Runnable() {
                    public void run() {
                        try {
                            Thread.sleep(delay);
                        } catch (Throwable e) {
                        }
                        doExport();
                    }
                });
                thread.setDaemon(true);
                thread.setName("DelayExportServiceThread");
                thread.start();
            } else {
                doExport();
            }
        }
    ```又调用了ServiceConfig的doExport方法，doExport调用doExportUrls，doExportUrls调用doExportUrlsFor1Protocol
    ，doExportUrlsFor1Protocol有如下代码片段
    
    
    
    
    <div class="se-preview-section-delimiter"></div>
    
    ```java
      if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
                    if (logger.isInfoEnabled()) {
                        logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                    }
                    if (registryURLs != null && registryURLs.size() > 0
                            && url.getParameter("register", true)) {
                        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()));
    
                            Exporter<?> exporter = protocol.export(invoker);
                            exporters.add(exporter);
                        }
                    } else {
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
    
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                }

重点在 Exporter

if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
                    if (logger.isInfoEnabled()) {
                        logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                    }
                    if (registryURLs != null && registryURLs.size() > 0
                            && url.getParameter("register", true)) {
                        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()));
    
                            Exporter<?> exporter = protocol.export(invoker);
                            exporters.add(exporter);
                        }
                    } else {
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
    
                        Exporter<?> exporter = protocol.export(invoker);
                        exporters.add(exporter);
                    }
                }

重点在 protocol.export(invoker);

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

默认会进入DubboProtocol

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;
        }

重点在 openServer(url)

private void openServer(URL url) {
            // find server.
            String key = url.getAddress();
            //client 也可以暴露一个只有server可以调用的服务。
            boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true);
            if (isServer) {
                ExchangeServer server = serverMap.get(key);
                if (server == null) {
                    serverMap.put(key, createServer(url));
                } else {
                    //server支持reset,配合override功能使用
                    server.reset(url);
                }
            }
        }

openServer又调用了createServer

private ExchangeServer createServer(URL url) {
            //默认开启server关闭时发送readonly事件
            url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
            //默认开启heartbeat
            url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
            String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
    
            if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
                throw new RpcException("Unsupported server type: " + str + ", url: " + url);
    
            url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
            ExchangeServer server;
            try {
                server = Exchangers.bind(url, requestHandler);
            } catch (RemotingException e) {
                throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
            }
            str = url.getParameter(Constants.CLIENT_KEY);
            if (str != null && str.length() > 0) {
                Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
                if (!supportedTypes.contains(str)) {
                    throw new RpcException("Unsupported client type: " + str);
                }
            }
            return server;
        }

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

### Exchangers ###

public static ExchangeServer bind(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).bind(url, handler);
        }

Exchangers 调用了HeaderExchanger的bind方法

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

重点看Transporters.bind方法

### Transporters ###

public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
            if (url == null) {
                throw new IllegalArgumentException("url == null");
            }
            if (handlers == null || handlers.length == 0) {
                throw new IllegalArgumentException("handlers == null");
            }
            ChannelHandler handler;
            if (handlers.length == 1) {
                handler = handlers[0];
            } else {
                handler = new ChannelHandlerDispatcher(handlers);
            }
            return getTransporter().bind(url, handler);
        }

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

默认会进入NettyTransporter的bind方法

public class NettyTransporter implements Transporter { 
    
        public static final String NAME = "netty";
    
        public Server bind(URL url, ChannelHandler listener) throws RemotingException {
            return new NettyServer(url, listener);
        }
    
        public Client connect(URL url, ChannelHandler listener) throws RemotingException {
            return new NettyClient(url, listener);
        }
    
    }

### NettyServer ###

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

public NettyServer(URL url, ChannelHandler handler) throws RemotingException{
            super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
        }

public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
            super(url, handler);
            localAddress = getUrl().toInetSocketAddress();
            String host = url.getParameter(Constants.ANYHOST_KEY, false) 
                            || NetUtils.isInvalidLocalHost(getUrl().getHost()) 
                            ? NetUtils.ANYHOST : getUrl().getHost();
            bindAddress = new InetSocketAddress(host, getUrl().getPort());
            this.accepts = url.getParameter(Constants.ACCEPTS_KEY, Constants.DEFAULT_ACCEPTS);
            this.idleTimeout = url.getParameter(Constants.IDLE_TIMEOUT_KEY, Constants.DEFAULT_IDLE_TIMEOUT);
            try {
                doOpen();
                if (logger.isInfoEnabled()) {
                    logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
                }
            } catch (Throwable t) {
                throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName() 
                                            + " on " + getLocalAddress() + ", cause: " + t.getMessage(), t);
            }
            if (handler instanceof WrappedChannelHandler ){
                executor = ((WrappedChannelHandler)handler).getExecutor();
            }
        }

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

@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);
    
            final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
            channels = nettyHandler.getChannels();
            // https://issues.jboss.org/browse/NETTY-365
            // https://issues.jboss.org/browse/NETTY-379
            // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
            bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
                public ChannelPipeline getPipeline() {
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
                    ChannelPipeline pipeline = Channels.pipeline();
                    /*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); }*/
                    pipeline.addLast("decoder", adapter.getDecoder());
                    pipeline.addLast("encoder", adapter.getEncoder());
                    pipeline.addLast("handler", nettyHandler);
                    return pipeline;
                }
            });
            // bind
            channel = bootstrap.bind(getBindAddress());
        }

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

[SouthEast]: /images/20220611/5739e1f0d3734b34b9d85bae88d39aa3.png
[SouthEast 1]: /images/20220611/7cbefeb3026e47d4a542d01c350dff83.png
[SouthEast 2]: /images/20220611/722496e65b3e404b9e7157795b9dfcaf.png