dubbo服务启动过程
2018-02-01 本文已影响3507人
一滴水的坚持
在项目启动过程中，dubbo服务如何随项目的启动而发布？dubbo如何随着spring容器的初始化而启动。
已知，在项目启动过程中，我们会将dubbo的配置文件写到spring的配置文件里，如下xml文件：
<dubbo:application name="anyname_provider" />
<!-- 使用zookeeper注册中心暴露服务地址 -->
<dubbo:registry address="zookeeper://127.0.0.1:2181" />
<!-- 用dubbo协议在20880端口暴露服务 -->
<dubbo:protocol name="dubbo" port="20880" />
<!-- 声明需要暴露的服务接口 -->
<dubbo:service interface="com.shxz130.provider.Provider"
ref="demoService" />
从官方文档中，我们能看到如下：
启动过程.png
也就是说spring启动过程中，随着Spring在初始化过程中，碰到dubbo命名的标签，如（<dubbo:service>,<dubbo:registry>）等标签，会由DubboNamespaceHandler类处理，具体原理见链接Spring自定义标签
DubboBeanDefinitionParser代码如下：
public class DubboNamespaceHandler extends NamespaceHandlerSupport {
static {
Version.checkDuplicate(DubboNamespaceHandler.class);
}
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 AnnotationBeanDefinitionParser());
}
}
遇到不同的标签，会由不同的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 {
}
该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服务。
部分ServiceBean的代码如下：
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();
}
}
//判断是否延迟发布
private boolean isDelay() {
Integer delay = getDelay();
ProviderConfig provider = getProvider();
if (delay == null && provider != null) {
delay = provider.getDelay();
}
return supportedApplicationListener && (delay == null || delay == -1);
}
//当bean初始化完成调用
public void afterPropertiesSet() throws Exception {
//......此处省略10000行代码
if (!isDelay()) {
//暴露服务
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);
}
}
这里以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
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);
}
对应官方文档中的设计，就联系起来了。
服务暴露
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服务就启动了。再回头看官方文档上的说明，就很清楚了。
暴露服务时序图.png
参考文档：http://blog.csdn.net/dachengxi/article/details/62567065
参考文档：http://blog.csdn.net/xcylive520/article/details/52347110
fyi
