标签:idea 忽略 find 打开 字段 channel abs rac cal
参考dubbo系列二、dubbo+zookeeper+dubboadmin分布式服务框架搭建(windows平台)
https://www.jianshu.com/p/16b72c10fca8
Spring自定义标签总共可以分为以下几个步骤
定义Bean 标签解析生成接收配置的POJO。
定义schema文件,定义自定义标签的attr属性
定义解析类parser,遇到自定义标签如何解析。
定义命名空间处理类namespaceSupport,遇到自定义的命名标签,能够路由到对应的解析类。
声明schema,写入spring.schema文件中
声明自定义标签的命名处理类namespaceHandler,写入spring.handlers文件中
例如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应用程序命名--> <dubbo:application name="dubbo-demo-provider"/> <!--dubbo注册地址--> <dubbo:registry address="zookeeper://192.168.1.100:2181"/> <!--dubbo协议地址--> <dubbo:protocol name="dubbo" port="20880"/> <!--接口声明--> <dubbo:service interface="com.dubbo.demo.api.DemoRpcService" ref="demoRpcService"/> <bean id="demoRpcService" class="com.dubbo.demo.DemoRpcServiceImpl"/> </beans>
官网:https://dubbo.incubator.apache.org/zh-cn/docs/dev/implementation.html
初始化过程细节
解析服务
基于 dubbo.jar 内的 META-INF/spring.handlers
配置,Spring 在遇到 dubbo 名称空间时,会回调 DubboNamespaceHandler
。
所有 dubbo 的标签,都统一用 DubboBeanDefinitionParser
进行解析,基于一对一属性映射,将 XML 标签解析为 Bean 对象。
在 ServiceConfig.export()
或 ReferenceConfig.get()
初始化时,将 Bean 对象转换 URL 格式,所有 Bean 属性转成 URL 的参数。
然后将 URL 传给 协议扩展点,基于扩展点的 扩展点自适应机制,根据 URL 的协议头,进行不同协议的服务暴露或引用。
public static void main(String[] args) throws IOException { ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("classpath:dubbo-provider.xml"); context.start(); // 阻塞当前进程,否则程序会直接停止 System.in.read(); }
spring启动过程中,随着Spring在初始化过程中,碰到dubbo命名的标签,如(<dubbo:service>
,<dubbo:registry>
)等标签,会由DubboNamespaceHandler
类处理,具体原理见链接Spring自定义标签
DubboNamespaceHandler类源码:
// // Source code recreated from a .class file by IntelliJ IDEA // (powered by Fernflower decompiler) // package com.alibaba.dubbo.config.spring.schema; 。。。import org.springframework.beans.factory.xml.NamespaceHandlerSupport; public class DubboNamespaceHandler extends NamespaceHandlerSupport { public DubboNamespaceHandler() { } public void init() {
// application标签解析 <dubbo:application name="dubbo-demo-provider"/> this.registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true)); // module标签解析
this.registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true)); // module标签解析
this.registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
this.registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
this.registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true)); this.registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
// <dubbo:protocol name="dubbo" port="20880"/>
this.registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
// service标签
// <dubbo:service interface="com.dubbo.demo.api.DemoRpcService" ref="demoRpcService"/>
// <bean id="demoRpcService" class="com.dubbo.demo.DemoRpcServiceImpl"/>
// </beans>
this.registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true)); this.registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false)); this.registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true)); } static { Version.checkDuplicate(DubboNamespaceHandler.class); } }
遇到不同的标签,会由不同的Parser
处理,这里重点看服务发布,这行代码:
registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
也就是说,当Spring容器处理完<dubbo:service>
标签后,会在Spring容器中生成一个ServiceBean
,服务的发布也会在ServiceBean
中完成。不妨看一下ServiceBean
的定义:
public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, BeanNameAware { }
ServiceBean 同时也是service标签解析之后的bean之一,继承ServiceConfig
该Bean实现了很多接口,关于InitializingBean
,DisposableBean
,ApplicationContextAware
,BeanNameAware
,这些接口的使用介绍如下链接:
而在Spring初始化完成Bean的组装,会调用InitializingBean
的afterPropertiesSet
方法,在Spring容器加载完成,会接收到事件ContextRefreshedEvent
,调用ApplicationListener
的onApplicationEvent
方法。afterPropertiesSet
中,和onApplicationEvent
中,会调用export()
,在export()
中,会暴露dubbo服务,具体区别在于是否配置了delay
属性,是否延迟暴露,如果delay
不为null
,或者不为-1
时,会在afterPropertiesSet
中调用export()
暴露dubbo服务,如果为null
,或者为-1
时,会在Spring容器初始化完成,接收到ContextRefreshedEvent
事件,调用onApplicationEvent
,暴露dubbo服务。public class ServiceBean<T> extends ServiceConfig<T> implements InitializingBean, DisposableBean, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent>, BeanNameAware { //Spring容器初始化完成,调用 public void onApplicationEvent(ContextRefreshedEvent event) { if (isDelay() && !isExported() && !isUnexported()) { if (logger.isInfoEnabled()) { logger.info("The service ready on spring started. service: " + getInterface()); } //暴露服务 export(); } } }
在export()
,暴露服务过程中,如果发现有delay属性
,则延迟delay时间
,暴露服务,如果没有,则直接暴露服务。
public synchronized void export() { //忽略若干行代码 if (delay != null && delay > 0) { //当delay不为null,且大于0时,延迟delay时间,暴露服务 delayExportExecutor.schedule(new Runnable() { public void run() { //暴露服务 doExport(); } }, delay, TimeUnit.MILLISECONDS); } else { //直接暴露服务 doExport(); } }
而在doExport()
中,验证参数,按照不同的Protocol
,比如(dubbo
,injvm
)暴露服务,在不同的zookeeper
集群节点上注册自己的服务。
protected synchronized void doExport() { //忽略10000行代码 doExportUrls(); //忽略10000行代码 } private void doExportUrls() { List<URL> registryURLs = loadRegistries(true); for (ProtocolConfig protocolConfig : protocols) { //按照不同的Protocal暴露服务 doExportUrlsFor1Protocol(protocolConfig, registryURLs); } }
// 获取注册中心地址 protected List<URL> loadRegistries(boolean provider) { checkRegistry(); List<URL> registryList = new ArrayList<URL>(); // protected List<RegistryConfig> registries;解析后的RegistryConfig中获取地址列表 if (registries != null && !registries.isEmpty()) { for (RegistryConfig config : registries) { String address = config.getAddress(); if (address == null || address.length() == 0) { address = Constants.ANYHOST_VALUE; } // 如果地址为空,再次从配置文件中取 String sysaddress = System.getProperty("dubbo.registry.address"); if (sysaddress != null && sysaddress.length() > 0) { address = sysaddress; } // 如果地址不为空,拼接协议类型、版本信息 if (address.length() > 0 && !RegistryConfig.NO_AVAILABLE.equalsIgnoreCase(address)) { Map<String, String> map = new HashMap<String, String>(); appendParameters(map, application); appendParameters(map, config); map.put("path", RegistryService.class.getName()); map.put("dubbo", Version.getProtocolVersion()); map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis())); if (ConfigUtils.getPid() > 0) { map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid())); } // 默认dubbo协议 if (!map.containsKey("protocol")) { if (ExtensionLoader.getExtensionLoader(RegistryFactory.class).hasExtension("remote")) { map.put("protocol", "remote"); } else { map.put("protocol", "dubbo"); } } // 如果同一个标签配置多个地址,则拆分 List<URL> urls = UrlUtils.parseURLs(address, map); for (URL url : urls) { url = url.addParameter(Constants.REGISTRY_KEY, url.getProtocol()); url = url.setProtocol(Constants.REGISTRY_PROTOCOL); if ((provider && url.getParameter(Constants.REGISTER_KEY, true)) || (!provider && url.getParameter(Constants.SUBSCRIBE_KEY, true))) { registryList.add(url); } } } } } // 返回格式化后的注册地址 return registryList; }
这里以dubbo
协议为例,看一下发布的过程,在发布过程中,会用一个变量map
保存URL的所有变量和value值,然后调用代理工程proxyFactory,获取代理类,然后将invoker转换成exporter,暴露服务,具体如下:
protocol://host:port/path?key=value&key=value
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { //如果协议类型为null,则默认为dubbo协议 String name = protocolConfig.getName(); if (name == null || name.length() == 0) { name = "dubbo"; } //map是保存url中key-Value的值 Map<String, String> map = new HashMap<String, String>(); //URL中的side属性,有两个值,一个provider,一个consumer,暴露服务的时候为provider map.put(Constants.SIDE_KEY, Constants.PROVIDER_SIDE); //dubbo的版本号 url中的dubbo map.put(Constants.DUBBO_VERSION_KEY, Version.getVersion()); //url中的timestamp map.put(Constants.TIMESTAMP_KEY, String.valueOf(System.currentTimeMillis())); //url中的pid if (ConfigUtils.getPid() > 0) { map.put(Constants.PID_KEY, String.valueOf(ConfigUtils.getPid())); } //从其他参数中获取参数 appendParameters(map, application); appendParameters(map, module); appendParameters(map, provider, Constants.DEFAULT_KEY); appendParameters(map, protocolConfig); appendParameters(map, this); //忽略若干代码 if (ProtocolUtils.isGeneric(generic)) { map.put("generic", generic); map.put("methods", Constants.ANY_VALUE); } else { //url中的revesion字段 String revision = Version.getVersion(interfaceClass, version); if (revision != null && revision.length() > 0) { map.put("revision", revision); } //拼接URL中的methods String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames(); if (methods.length == 0) { logger.warn("NO method found in service interface " + interfaceClass.getName()); map.put("methods", Constants.ANY_VALUE); } else { map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ",")); } } //token 临牌校验 if (!ConfigUtils.isEmpty(token)) { if (ConfigUtils.isDefault(token)) { map.put("token", UUID.randomUUID().toString()); } else { map.put("token", token); } } //injvm协议 if ("injvm".equals(protocolConfig.getName())) { protocolConfig.setRegister(false); map.put("notify", "false"); } //获取上下文路径 String contextPath = protocolConfig.getContextpath(); if ((contextPath == null || contextPath.length() == 0) && provider != null) { contextPath = provider.getContextpath(); } //获取主机名 String host = this.findConfigedHosts(protocolConfig, registryURLs, map); //获取端口 Integer port = this.findConfigedPorts(protocolConfig, name, map); //组装URL,将map中的协议,版本号信息等 URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map); //如果url使用的协议存在扩展,调用对应的扩展来修改原url if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class) .hasExtension(url.getProtocol())) { url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class) .getExtension(url.getProtocol()).getConfigurator(url).configure(url); } String scope = url.getParameter(Constants.SCOPE_KEY); //配置为none不暴露 if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { //如果不是remote,则暴露本地服务 exportLocal(url); } //如果配置不是local则暴露为远程服务 if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) { //忽略日志 如果注册中心地址不为null if (registryURLs != null && registryURLs.size() > 0) { for (URL registryURL : registryURLs) { url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic")); //忽略不相干的代码 // 通过代理工厂将ref对象转化成invoker对象 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); //代理invoker对象 DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); // 暴露服务 Exporter<?> exporter = protocol.export(wrapperInvoker); //一个服务可能有多个提供者,保存在一起 exporters.add(exporter); } } else { Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url); DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); Exporter<?> exporter = protocol.export(wrapperInvoker); exporters.add(exporter); } } } this.urls.add(url); }
将doExportUrlsFor1Protocol
代码再简化一下,如下:
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { Map map=builderUrl(); // 通过代理工厂将ref对象转化成invoker对象 Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url); //代理invoker对象 DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this); // 暴露服务 Exporter<?> exporter = protocol.export(wrapperInvoker); //一个服务可能有多个提供者,保存在一起 exporters.add(exporter); }
拼接后的url:
dubbo://192.168.1.100:20880/com.dubbo.demo.api.DemoRpcService?anyhost=true&application=dubbo-demo-provider&bind.ip=192.168.1.100&bi
nd.port=20880&dubbo=2.6.0&generic=false&interface=com.dubbo.demo.api.DemoRpcService&methods=getUserName&pid=18740&side=provider×tamp=1538311737815
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
而在上面proxyFactory.getInvoker
中,很显然是获取到的是接口的代理类。
而在 protocol.export(wrapperInvoker)
中,将服务暴露出去。
代码如下:
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { URL url = invoker.getUrl(); //忽略若干代码 //打开服务 openServer(url); optimizeSerialization(url); return exporter; }
private void openServer(URL url) { String key = url.getAddress(); boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true); //是否server端 if (isServer) { ExchangeServer server = serverMap.get(key); if (server == null) { //如果服务不存在,创建服务 serverMap.put(key, createServer(url)); } else { server.reset(url); } } }
private ExchangeServer createServer(URL url) { //忽略若干代码 ExchangeServer server; try { server = Exchangers.bind(url, requestHandler); } catch (RemotingException e) { throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e); } return server; }
而在headerExchanger
的bind
中,调用了Transporters.bind()
,一直调用到NettyServer
,绑定了端口和链接。
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); } //Transporters.bind public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException { //忽略很多代码 return getTransporter().bind(url, handler); } //上段代码的getTransporter() public static Transporter getTransporter() { return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension(); }
而在Transporter的定义中看到下面代码:
@SPI("netty") public interface Transporter { @Adaptive({Constants.SERVER_KEY, Constants.TRANSPORTER_KEY}) Server bind(URL url, ChannelHandler handler) throws RemotingException; @Adaptive({Constants.CLIENT_KEY, Constants.TRANSPORTER_KEY}) Client connect(URL url, ChannelHandler handler) throws RemotingException; }
所以这里调用的是NettyTransporter
,这里启动了一个新的NettyServer
。
public class NettyTransporter implements Transporter { public static final String NAME = "netty4"; 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的构造方法中,调用了父类的构造方法,调用了doOpen()
方法指定了端口
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 AbstractServer(URL url, ChannelHandler handler) throws RemotingException { super(url, handler); //忽略很多代码 doOpen(); //忽略很多代码 } @Override protected void doOpen() throws Throwable { NettyHelper.setNettyLoggerFactory(); bootstrap = new ServerBootstrap(); bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), new DefaultThreadFactory("NettyServerWorker", true)); final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); channels = nettyServerHandler.getChannels(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE) .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) .childHandler(new ChannelInitializer<NioSocketChannel>() { @Override protected void initChannel(NioSocketChannel ch) throws Exception { NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug .addLast("decoder", adapter.getDecoder()) .addLast("encoder", adapter.getEncoder()) .addLast("handler", nettyServerHandler); } }); // bind ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); channelFuture.syncUninterruptibly(); channel = channelFuture.channel(); }
到这里dubbo服务就启动了,但是有一点还是有疑惑,那么,dubbo服务什么时候注册到注册中心的?带着疑惑看了一下官方文档。
也就是说,在调用DubboProtocol
暴露服务之前,回去调用拦截器,当发现是regiester
,则去注册中心注册服务。
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { //如果是registerProtocol,则调用RegisterProtocol.export方法 if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { return protocol.export(invoker); } return new ListenerExporterWrapper<T>(protocol.export(invoker), Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class) .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY))); }
而在RegisterProtocol.export
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { //export invoker final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker); URL registryUrl = getRegistryUrl(originInvoker); //根据SPI机制获取具体的Registry实例,这里获取到的是ZookeeperRegistry final Registry registry = getRegistry(originInvoker); final URL registedProviderUrl = getRegistedProviderUrl(originInvoker); boolean register = registedProviderUrl.getParameter("register", true); if (register) { //在这里注册服务 register(registryUrl, registedProviderUrl); //忽略很多代码 } //忽略很多代码 } public void register(URL registryUrl, URL registedProviderUrl) { Registry registry = registryFactory.getRegistry(registryUrl); registry.register(registedProviderUrl); }
而ZookeeperRegistry
继承父类FailbackRegistry
,在父类的register
方法中,调用了 doRegister
,doRegister
中,创建了ZK节点,这样就将自己的服务暴露到注册中心zk上:
@Override public void register(URL url) { //忽略很多代码 doRegister(url); //忽略很多代码 } protected void doRegister(URL url) { try { zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true)); } catch (Throwable e) { throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); } }
这样,整个dubbo服务就启动了。再回头看官方文档上的说明,就很清楚了。
标签:idea 忽略 find 打开 字段 channel abs rac cal
原文地址:https://www.cnblogs.com/wangzhuxing/p/9733339.html