背景
服务暴露网上已经有很多文章了，大而全，我们这里主要抓细节😄。
image
疑问
暴露过程做了些啥？
是先启动服务还是先连接注册中心？
服务下线怎么感知注册中心？
暴露
我们从 org.apache.dubbo.config.ServiceConfig#doExportUrls() 方法进去
private void doExportUrls() {
List<URL> registryURLs = loadRegistries(true);
//支持多协议暴露就是说 <dubbo:protocol 可以多个
//<dubbo:protocol name="dubbo" port="20880"/>
//<dubbo:protocol name="rest" port="20881"/>
//像这样，如果有php客户端 和 dubbo客户端都可以同事支持
for (ProtocolConfig protocolConfig : protocols) {
doExportUrlsFor1Protocol(protocolConfig, registryURLs);
}
}
进入 doExportUrlsFor1Protocol()中，这个方法大家一定要进去瞅一眼，和我们写的代码也差不多，方法长度太长，而且循环嵌套很深。
//org.apache.dubbo.config.ServiceConfig#doExportUrlsFor1Protocol
private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {
String name = protocolConfig.getName();
if (StringUtils.isEmpty(name)) {//没有配置协议，默认dubbo
name = DUBBO;
}
Map<String, String> map = new HashMap<String, String>();
map.put(SIDE_KEY, PROVIDER_SIDE);
//将所有的配置都放到URL 的key=value 中
appendRuntimeParameters(map);
appendParameters(map, metrics);
appendParameters(map, application);
appendParameters(map, module);
appendParameters(map, provider);
appendParameters(map, protocolConfig);
appendParameters(map, this);
if (ProtocolUtils.isGeneric(generic)) {//泛化
map.put(GENERIC_KEY, generic);
map.put(METHODS_KEY, ANY_VALUE);
} else {//版本
String revision = Version.getVersion(interfaceClass, version);
if (revision != null && revision.length() > 0) {
map.put(REVISION_KEY, revision);
}
String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
if (methods.length == 0) {
logger.warn("No method found in service interface " + interfaceClass.getName());
map.put(METHODS_KEY, ANY_VALUE);
} else {
map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
}
}
//token,dubbo 支持token校验，只有携带对的token才能调用成功
if (!ConfigUtils.isEmpty(token)) {
if (ConfigUtils.isDefault(token)) {
map.put(TOKEN_KEY, UUID.randomUUID().toString());
} else {
map.put(TOKEN_KEY, token);
}
}
String host = this.findConfigedHosts(protocolConfig, registryURLs, map);
Integer port = this.findConfigedPorts(protocolConfig, name, map);
URL url = new URL(name, host, port, getContextPath(protocolConfig).map(p -> p + "/" + path).orElse(path), map);
if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.hasExtension(url.getProtocol())) {
url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.getExtension(url.getProtocol()).getConfigurator(url).configure(url);
}
String scope = url.getParameter(SCOPE_KEY);
// don't export when none is configured
if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
exportLocal(url);//先将服务暴露到本地，下面分析
}
// export to remote if the config is not local (export to local only when config is local)
if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
if (CollectionUtils.isNotEmpty(registryURLs)) {
//注册中心也支持多个，比如可以将服务暴露到集群内，也可以将
//服务暴露到中台供所其他业务线用
for (URL registryURL : registryURLs) {
//if protocol is only injvm ,not register
if (LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
continue;
}
url = url.addParameterIfAbsent(DYNAMIC_KEY, registryURL.getParameter(DYNAMIC_KEY));
//加载监控配置
URL monitorUrl = loadMonitor(registryURL);
if (monitorUrl != null) {
url = url.addParameterAndEncoded(MONITOR_KEY, monitorUrl.toFullString());
}
// 调用具体bean的代理模式，默认为javassist
String proxy = url.getParameter(PROXY_KEY);
if (StringUtils.isNotEmpty(proxy)) {
registryURL = registryURL.addParameter(PROXY_KEY, proxy);
}
//组装invoker
Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
//暴露服务
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
}
} else {
Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
}
//存储发布信息
MetadataReportService metadataReportService = null;
if ((metadataReportService = getMetadataReportService()) != null) {
metadataReportService.publishProvider(url);
}
}
}
this.urls.add(url);
}
本地暴露 exportLocal(url)
//org.apache.dubbo.config.ServiceConfig#exportLocal
private void exportLocal(URL url) {
URL local = URLBuilder.from(url)
.setProtocol(LOCAL_PROTOCOL)//收到设置协议为injvm，以供下面选择对应的protocol
.setHost(LOCALHOST_VALUE)
.setPort(0)
.build();
//
Exporter<?> exporter = protocol.export(
PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, local));
exporters.add(exporter);
}
static Protocol protocol = ExtensionLoader
.getExtensionLoader(Protocol.class).getAdaptiveExtension();
static ProxyFactory PROXY_FACTORY = ExtensionLoader
.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();
protocol 静态变量为 Protocol 接口的自适应扩展点，调用 protocol.export(Invoker<T> invoker) 将会根据传入的invoker 信息决定去往哪个实现类。而 invoker 传入的值为
PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, local) ，PROXY_FACTORY 静态变量也是一个 ProxyFactory 的扩展点，从下面可以看到该扩展点为方法扩展点，这里我们并没有个自定义过proxy属性，默认实现为javassist=JavassistProxyFactory；（这里忽略各种包装器）
@SPI("javassist")
public interface ProxyFactory {
@Adaptive({"proxy"})
<T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) ;
}
进到JavassistProxyFactory 的 getInvoker实现中。
//org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory#getInvoker
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
//这里的proxy 是我们真正的实现类HelloServiceImpl@xxx,
//如果传进来的是一个代理类实现的花，这里只取接口type=HelloService
Class cls = proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type;
//将 HelloServiceImpl包装成一个Wrapper类，而wrapper对象的创建方式正式默认的javassist
final Wrapper wrapper = Wrapper.getWrapper(cls);
//返回一个匿名内部类对象，对象 doInvoke 方法中持有wrapper对象
//AbstractProxyInvoker 实现了Invoker
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
}
上面这种匿名的写法可能不够具体，我们通过自定义类的方式去实现它，更具象点
//org.apache.dubbo.rpc.proxy.javassist.JavassistProxyFactory#getInvoker
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
return new MyProxyInvoker(proxy,type,url,wrapper);
}
public class MyProxyInvoker extends AbstractProxyInvoker {
private Wrapper wrapper;
public MyProxyInvoker(Object proxy, Class type, URL url, Wrapper wrapper) {
super(proxy, type, url);
this.wrapper = wrapper;
}
@Override
protected Object doInvoke(Object proxy, String methodName, Class[] parameterTypes, Object[] arguments) throws Throwable {
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
}
这样写的效果是一样的 JavassistProxyFactory#getInvoker()方法返回的是 MyProxyInvoker 对象，后面我们就用该对象来描述分析。
回到 Exporter<?> exporter = protocol.export(
PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, local));中，表达式变成了Exporter<?> exporter = protocol.export(MyProxyInvoker)，MyProxyInvoker中的url对象为local
URL local = URLBuilder.from(url)
.setProtocol("injvm")
.setHost(LOCALHOST_VALUE)
.setPort(0)
.build();
所以protocol.export()的实现类为InjvmProtocol
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(),
exporterMap);
}
该方法返回 InjvmExporter，最后执行 exporters.add(exporter)，将InjvmExporter（这里其实外面会包装一层ListenerExporterWrapper包装器） 对象暴露到map中结束了jvm本地暴露。
远程暴露
我们再来看看远程暴露的区别
//同本地暴露一样返回MyProxyInvoker实例
Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(EXPORT_KEY, url.toFullString()));
//区别本地暴露 将 MyProxyInvoker实例 包装为 DelegateProviderMetaDataInvoker
DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
//这里写法和本地暴露一样，区别在于 wrapperInvoker 中的url#protocol 并不是injvm
Exporter<?> exporter = protocol.export(wrapperInvoker);
exporters.add(exporter);
我们dubug 看看 wrapperInvoker 中的url#protocol 是啥
image
Protocol 为registry，所以流程会进入到 RegistryProtocol#export（同样这里也会有Wrapper包装）我们debug进去，这个方法内容太丰富了，这里我们先只分析服务暴露
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
URL registryUrl = getRegistryUrl(originInvoker);
// 获取要暴露到注册中心的url
URL providerUrl = getProviderUrl(originInvoker);
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
//暴露服务 下面分析
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker, providerUrl);
//...
return new DestroyableExporter<>(exporter);
}
暴露服务 doLocalExport()
//org.apache.dubbo.registry.integration.RegistryProtocol#doLocalExport
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker, URL providerUrl) {
//将要暴露的服务生成唯一的key，避免重复
String key = getCacheKey(originInvoker);
//再次包装invoker，然后暴露
return (ExporterChangeableWrapper<T>) bounds.computeIfAbsent(key, s -> {
//providerUrl 为dubbo://xxx
Invoker<?> invokerDelegate = new InvokerDelegate<>(originInvoker, providerUrl);
//protocol.export 经过各种Wrapper 会进入到Dubbo.export
return new ExporterChangeableWrapper<>((Exporter<T>) protocol.export(invokerDelegate), originInvoker);
});
}
private String getCacheKey(final Invoker<?> originInvoker) {
URL providerUrl = getProviderUrl(originInvoker);
String key = providerUrl.removeParameters("dynamic", "enabled").toFullString();
return key;
}
这里originInvoker为 DelegateProviderMetaDataInvoker(MyProxyInvoker(HelloServiceImpl@xx))，invokerDelegate再次包装为InvokerDelegate(DelegateProviderMetaDataInvoker(MyProxyInvoker(HelloServiceImpl@xx)))，我们继续debug，到了ProtocolFilterWrapper#export
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
if (REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
return protocol.export(buildInvokerChain(invoker, SERVICE_FILTER_KEY, CommonConstants.PROVIDER));
}
private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) {
Invoker<T> last = invoker;
List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
if (!filters.isEmpty()) {
for (int i = filters.size() - 1; i >= 0; i--) {
//.....
};
}
}
return new CallbackRegistrationInvoker<>(last, filters);
}
buildInvokerChain()会将InvokerDelegate关联多个Filter过滤器，然后包装为CallbackRegistrationInvoker对象返回，我们接着debug，最后到了DubboProtocol#export()，此时的invoker为CallbackRegistrationInvoker(InvokerDelegate(DelegateProviderMetaDataInvoker(MyProxyInvoker(HelloServiceImpl@xx))))
image
//org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#export
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
URL url = invoker.getUrl();
//生成服务key=com.poizon.study.api.service.HelloService:20880，和方法无关
String key = serviceKey(url);
//将CallbackRegistrationInvoker包装为DubboExporter，然后存储在map中
//这个map 很关键，将作为后面调用寻找服务的入口
DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
exporterMap.put(key, exporter);
//export an stub service for dispatching event
Boolean isStubSupportEvent = url.getParameter(STUB_EVENT_KEY, DEFAULT_STUB_EVENT);
Boolean isCallbackservice = url.getParameter(IS_CALLBACK_SERVICE, false);
if (isStubSupportEvent && !isCallbackservice) {
String stubServiceMethods = url.getParameter(STUB_EVENT_METHODS_KEY);
if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
if (logger.isWarnEnabled()) {
logger.warn(new IllegalStateException("consumer [" + url.getParameter(INTERFACE_KEY) +
"], has set stubproxy support event ,but no stub methods founded."));
}
} else {
stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
}
}
//开启服务，也就是调用netty，开启20880端口
openServer(url);
//加载指定序列化方式 默认采用hessan2
optimizeSerialization(url);
return exporter;
}
//org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#openServer
private void openServer(URL url) {
//..... createServer()创建服务
serverMap.put(key, createServer(url));
}
//org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#createServer
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;
}
//org.apache.dubbo.remoting.exchange.support.header.HeaderExchanger#bind
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
//org....remoting.Transporters#bind(URL, ChannelHandler...)
public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException {
ChannelHandler handler;
if (handlers.length == 1) {
handler = handlers[0];
} else {
handler = new ChannelHandlerDispatcher(handlers);
}//默认选择netty4 实现
return getTransporter().bind(url, handler);
}
//org.apache.dubbo.remoting.transport.netty4.NettyTransporter#connect
public Client connect(URL url, ChannelHandler listener) throws RemotingException {
return new NettyClient(url, listener);
}
//org.apache.dubbo.remoting.transport.AbstractClient#AbstractClient
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
super(url, handler);
needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
try {
doOpen();
//org.apache.dubbo.remoting.transport.netty4.NettyClient#doOpen
protected void doOpen() throws Throwable {
final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
bootstrap = new Bootstrap();
bootstrap.group(nioEventLoopGroup)
.option(ChannelOption.SO_KEEPALIVE, true)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
.channel(NioSocketChannel.class);
if (getConnectTimeout() < 3000) {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
} else {
bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
}
bootstrap.handler(new ChannelInitializer() {
@Override
protected void initChannel(Channel ch) throws Exception {
跟到最后看到了熟悉的netty启动，这里有好多我们熟悉的配置，比如第一篇文章我们说到的心跳实现IdleStateHandler，以及心跳默认时间 UrlUtils.getHeartbeat(getUrl())，还有netty 的自定义handler nettyClientHandler(没错这个handler就是处理dubbo消费者请求的)
总结
总结下，我们一根线走到底，走到了最后的socket启动，最后将 DubboExporter 放入了map中，最后层层包装为 DestroyableExporter(ExporterChangeableWrapper(ListenerExporterWrapper(DubboExporter(CallbackRegistrationInvoker(InvokerDelegate(DelegateProviderMetaDataInvoker(MyProxyInvoker(HelloServiceImpl@xx))))))))；嵌套虽然多了点，但是Wrapper 类的功能都是为了扩展小功能，后面我们调几个分析
image
后面将分析注册中心和Wrapper 等功能。
网友评论