dubbo原理系列1-服务端暴露过程
2017-07-29 18:11
423 查看
基础要求:
1 使用dubbo开发过项目
2 了解spring的NamespaceHandler
3 了解一些Netty基本API
Now,进入主题吧。
** 处替换成你自己的服务接口完整路径哦。
Main会执行SpringContainer的start方法
可见只是用ClassPathXmlApplicationContext来触发spring容器启动。
故,我们可以推测出另一种方式便是通过WebApplicationContext
注意ServiceBean和AnnotationBean
这两个bean都继承了ServiceConfig,且ServiceBean 实现了ApplicationListener接口,我们知道容器启动时,会触发onApplicationEvent方法
重点在 Exporter
重点在 protocol.export(invoker);
默认会进入DubboProtocol
重点在 openServer(url)
openServer又调用了createServer
createServer又调用了server = Exchangers.bind(url, requestHandler);
Exchangers 调用了HeaderExchanger的bind方法
重点看Transporters.bind方法
默认会进入NettyTransporter的bind方法
可以看到父类构造方法中调用了doOpen方法后就输出了Start的日志,可见doOpen方法中就是启动了服务端,做好接受调用准备
实际也就是这样,doOpen方法利用Netty开启了服务端的监听,并绑定到InetSocketAddress(getBindAddress()方法返回),这个对象中包含了服务的地址端口。
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) { } } } }
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);
默认会进入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); }
默认会进入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()方法返回),这个对象中包含了服务的地址端口。
相关文章推荐
- dubbo原理系列1-服务端暴露过程
- dubbo原理系列3-consumer调用过程
- dubbo原理系列3-consumer调用过程
- dubbo原理系列2-reference代理生成过程
- dubbo原理系列2-reference代理生成过程
- 自定义View Layout过程 - 最易懂的自定义View原理系列(3)
- Dubbo学习过程、使用经验分享及实现原理简单介绍
- dubbo暴露服务过程
- 自定义View Draw过程- 最易懂的自定义View原理系列(4)
- 自定义View Layout过程 - 最易懂的自定义View原理系列(3)
- dubbo服务暴露过程源码分析
- 一起学DNS系列(十)图、例详解DNS递归和迭代查询原理及过程 (2)
- dubbo学习过程、使用经验分享及实现原理简单介绍
- 【Dubbo基础】dubbo学习过程、使用经验分享及实现原理简单介绍
- dubbo的服务暴露原理
- 网络编程中客户端和服务端通信过程、原理及代码示例
- DNS系列教程(十)图例详解DNS递归和迭代查询原理及过程 (2)
- Dubbo学习过程、使用经验分享及实现原理简单介绍
- 自定义View Draw过程- 最易懂的自定义View原理系列(4)
- 纯ant命令行打包android apk之图文从原理角度完全详解android打包过程(打包系列教程之一)