Dubbo之服务暴露源码分析
2018-03-29 本文已影响49人
土豆肉丝盖浇饭
时序图
在讲解源码前，先看下官方文档提供的时序图，后面的讲解基本是这个路线，但是会更细节化
服务暴露.png
大致逻辑
首先服务的实现bean在我们的spring容器中，我们会创建一个Invoker通过代理调用ref中的方法，同时Invoker会在protocol的export方法中会转换为Exporter，并且保存在protocol对象的exporterMap中，然后进行暴露。
重要概念
Protocol
Protocol 是服务域，它是 Invoker 暴露和引用的主功能入口，它负责 Invoker
的生命周期管理。
接口定义如下
@SPI("dubbo")
public interface Protocol {
/**
* 获取缺省端口，当用户没有配置端口时使用。
*
* @return 缺省端口
*/
int getDefaultPort();
/**
* 暴露远程服务：<br>
* 1. 协议在接收请求时，应记录请求来源方地址信息：RpcContext.getContext().setRemoteAddress();<br>
* 2. export()必须是幂等的，也就是暴露同一个URL的Invoker两次，和暴露一次没有区别。<br>
* 3. export()传入的Invoker由框架实现并传入，协议不需要关心。<br>
*
* @param <T> 服务的类型
* @param invoker 服务的执行体
* @return exporter 暴露服务的引用，用于取消暴露
* @throws RpcException 当暴露服务出错时抛出，比如端口已占用
*/
@Adaptive
<T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
/**
* 引用远程服务：<br>
* 1. 当用户调用refer()所返回的Invoker对象的invoke()方法时，协议需相应执行同URL远端export()传入的Invoker对象的invoke()方法。<br>
* 2. refer()返回的Invoker由协议实现，协议通常需要在此Invoker中发送远程请求。<br>
* 3. 当url中有设置check=false时，连接失败不能抛出异常，并内部自动恢复。<br>
*
* @param <T> 服务的类型
* @param type 服务的类型
* @param url 远程服务的URL地址
* @return invoker 服务的本地代理
* @throws RpcException 当连接服务提供方失败时抛出
*/
@Adaptive
<T> Invoker<T> refer(Class<T> type, URL url) throws RpcException;
/**
* 释放协议：<br>
* 1. 取消该协议所有已经暴露和引用的服务。<br>
* 2. 释放协议所占用的所有资源，比如连接和端口。<br>
* 3. 协议在释放后，依然能暴露和引用新的服务。<br>
*/
void destroy();
}
Invoker
Invoker 是实体域，它是 Dubbo 的核心模型，其它模型都向它靠扰，或转换成
它，它代表一个可执行体，可向它发起 invoke 调用，它有可能是一个本地的
实现，也可能是一个远程的实现，也可能一个集群实现。
接口定义如下
public interface Invoker<T> extends Node {
Class<T> getInterface();
Result invoke(Invocation invocation) throws RpcException;
}
Invocation
Invocation 是会话域，它持有调用过程中的变量，比如方法名，参数等。
接口定义如下
public interface Invocation {
String getMethodName();
Class<?>[] getParameterTypes();
Object[] getArguments();
Map<String, String> getAttachments();
String getAttachment(String key);
String getAttachment(String key, String defaultValue);
Invoker<?> getInvoker();
}
Exporter
Exporter用来封装不同协议暴露的Invoker，因为Invoker可以被多个Protocol暴露，因为每种Protocol都有各自的Exproter子类
接口定义如下
public interface Exporter<T> {
Invoker<T> getInvoker();
void unexport();
}
我的一些定义
本地暴露
本地暴露分为两种，通过远程协议还是本地协议暴露
本地协议的话，基于进程通信，所以不需要进行远程暴露，具体实现只有InjvmProtocol
而基于远程协议的暴露，需要开启服务监听，处理其他进程发来的rpc请求，同时可以选择进行远程暴露，具体实现有DubboProtocol，HessianProtocol等
远程暴露
远程暴露，就是将本地暴露的url发布到注册中心，这个暴露为了让服务引用者感知到服务的存在
远程暴露对应RegistryProtocol
远程暴露URL和本地暴露URL
进行远程暴露的时候，要先进行本地暴露，所以远程暴露URL里面有一个export参数会包含本地暴露URL
远程暴露URL主要是用来选择暴露的注册中心，注册本地暴露URL，以及增加事件监听
源码分析
解析配置
先看下我们平时是如何配置dubbo服务暴露的
<dubbo:service interface="com.alibaba.dubbo.demo.bid.BidService" ref="bidService" protocol="dubbo" />
上面的配置会通过自定义解析器DubboNamespaceHandler解析到ServiceBean对象
registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
然后在ServiceBean初始化完成后进行服务暴露
image.png
可以看到ServiceBean实现了Initializing接口，可以在afterPropertiesSet看到服务暴露的逻辑
if (! isDelay()) {
export();
}
这边的Delay并不是服务具体的暴露行为进行延迟，而是控制这个暴露行为在什么时候触发
private boolean isDelay() {
Integer delay = getDelay();
ProviderConfig provider = getProvider();
if (delay == null && provider != null) {
delay = provider.getDelay();
}
return supportedApplicationListener && (delay == null || delay.intValue() == -1);
}
上述代码的意思是，如果支持Spring的事件监听，并且没有配置延迟暴露，推迟到容器refresh完成的时候触发服务暴露逻辑，如果配置了delay,那么直接在afterPropertiesSet内调用暴露方法
我理解为一个是容器级别的delay，一个是服务级别的delay
export方法
export方法在ServiceConfig中
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();
}
}
这边会根据是否配置了delay参数，进行延迟暴露，通过线程休眠来实现
doExport方法涉及很多参数的校验与设置，遇到具体功能点再做分析，具体暴露逻辑调用了方法doExportUrls
private void doExportUrls() {
List<URL> registryURLs = loadRegistries(true);
for (ProtocolConfig protocolConfig : protocols) {
doExportUrlsFor1Protocol(protocolConfig, registryURLs);
}
}
doExportUrls方法首先会获取注册中心的URL，虽说可以配置很多个注册中心，但是我们就把它当成一个好了
然后根据service配置的不同协议，调用doExportUrlsFor1Protocol方法分别进行暴露
在doExportUrlsFor1Protocol的前半部分又是各种参数的提取，用来生成最终暴露的URL，我们关注核心的暴露逻辑
//1
URL url = new URL(name, host, port, (contextPath == null || contextPath.length() == 0 ? "" : contextPath + "/") + path, map);
if (ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.hasExtension(url.getProtocol())) {
url = ExtensionLoader.getExtensionLoader(ConfiguratorFactory.class)
.getExtension(url.getProtocol()).getConfigurator(url).configure(url);
}
//2
String scope = url.getParameter(Constants.SCOPE_KEY);
//配置为none不暴露
if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {
//配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)
//3
if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {
exportLocal(url);
}
//如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)
//4
if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){
if (logger.isInfoEnabled()) {
logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
}
//5
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);
}
}
}
this.urls.add(url);
注意上面代码1处，就是使用之前提取的参数生成本地暴露url的逻辑
而代码2处的scope也还是比较重要的，它控制了服务应该怎么暴露，我们项目中一般对service不进行scope配置，那么取到的值为null，代码3和代码4的条件都会满足，既会进行本地协议的本地暴露，也会进行远程暴露
而代码5，我们可以配置register="false",直接进行远程协议的本地暴露，不记录到注册中心上去，但是我们还是可以通过在消费者强制配置url来调用
<dubbo:service interface="com.alibaba.dubbo.demo.bid.BidService" ref="bidService" protocol="dubbo" register="false"/>
服务暴露的逻辑其实是同一套
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
Exporter<?> exporter = protocol.export(invoker);
主要的区别点还是在于Url的不同，因为url带了不同的protocol以及其他配置，然后具体暴露时，使用之前讲的SPI来调用不同实现
比如在exportLocal方法里，其实会把url的protocol修改为injvm
private void exportLocal(URL url) {
if (!Constants.LOCAL_PROTOCOL.equalsIgnoreCase(url.getProtocol())) {
//修改protocol为Injvm
URL local = URL.valueOf(url.toFullString())
.setProtocol(Constants.LOCAL_PROTOCOL)
.setHost(NetUtils.LOCALHOST)
.setPort(0);
// modified by lishen
ServiceClassHolder.getInstance().pushServiceClass(getServiceClass(ref));
Exporter<?> exporter = protocol.export(
proxyFactory.getInvoker(ref, (Class) interfaceClass, local));
exporters.add(exporter);
logger.info("Export dubbo service " + interfaceClass.getName() +" to local registry");
}
}
在存在注册中心，并且服务的Registry属性不为false的情况下会进行远程暴露，会在注册中心url的export参数带上原先的本地暴露url进行远程暴露，因此暴露使用的protocol也相应变为RegistryProtocol
Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
下面讲解具体暴露的逻辑
ref转换为Invoker
在进行暴露之前，我们需要将spring容器内的接口实现ref转换为invoker，通过proxyFactory.getInvoker(ref, (Class) interfaceClass, local)方法
proxyFactory是一个扩展点，有javaassist和jdk动态代理两种实现，默认实现为javaassist，并且提供一个包装类StubProxyFactoryWrapper用于提供降级服务(以后单独讲解)
public class JavassistProxyFactory extends AbstractProxyFactory {
@SuppressWarnings("unchecked")
public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// TODO Wrapper类不能正确处理带$的类名
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
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);
}
};
}
}
JavassistProxyFactory 中的Wapper类是动态生成的，可以针对接口的每个方法生成直接调用的代码，避免了反射，因为做了缓存，多次调用的情况下，会加快效率，而jdk实现用的反射效率应该差多了
//JdkProxyFactory的AbstractProxyInvoker实现
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
Method method = proxy.getClass().getMethod(methodName, parameterTypes);
return method.invoke(proxy, arguments);
}
};
通过proxyFactory我们会得到一个父类为AbstractProxyInvoker的匿名Invoker类，内部通过反射或者动态生成字节码来调用目标ref的方法
通过protocol暴露
在得到Invoker后，我们通过protocol去进行服务暴露，暴露成功后得到Exporter引用
<T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
通过Protocol的export方法，需要将invoker转换为exporter，为什么？
因为Invoker只负责对具体方法的调用，但是方法的调用可以暴露到多个Protocol，所以需要有具体的Exporter来对应,比如Dubbo暴露得到DubboExporter，injvm暴露得到InjvmExporter
下面讲解每种protocol的暴露
通过InjvmProtocol暴露
InjvmProtocol是本地暴露中唯一使用本地协议的，意思就是说这个服务的url不能发布到注册中心，只能本地消费，在dubbo引用服务的逻辑中，如果发现本地InjvmProtocol中有所需要的Exproter，会优先选择引用本地
看下暴露的代码
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
return new InjvmExporter<T>(invoker, invoker.getUrl().getServiceKey(), exporterMap);
}
exporterMap为InjvmProtocol继承AbstractProtocol的一个参数，用来保存Exproter引用，同时exporterMap也会在InjvmExporter内被引用，主要用于卸载功能
class InjvmExporter<T> extends AbstractExporter<T> {
private final String key;
private final Map<String, Exporter<?>> exporterMap;
InjvmExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap){
super(invoker);
this.key = key;
this.exporterMap = exporterMap;
exporterMap.put(key, this);
}
public void unexport() {
super.unexport();
exporterMap.remove(key);
}
}
关于这个exporterMap，由于每个Protocol实现都继承了AbstractProtocol，所以都会有exporterMap属性，并且每种Protocol在容器内只存在一个，我们可以在每个Protocol对象的exporterMap中拿到这个Protocol暴露的所有Exproter
通过RegistryProtocol暴露
这是远程暴露，在进行本地暴露的同时将本地暴露的url注册到注册中心同时也注册事件监听
public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {
//export invoker
//1
final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
//registry provider
final Registry registry = getRegistry(originInvoker);
final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
registry.register(registedProviderUrl);
// 订阅override数据
// FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。
final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
//注册监听事件，用于url被修改时回调，进行exporter重新暴露
registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
//保证每次export都返回一个新的exporter实例
return new Exporter<T>() {
public Invoker<T> getInvoker() {
return exporter.getInvoker();
}
public void unexport() {
try {
exporter.unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
try {
registry.unregister(registedProviderUrl);
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
try {
overrideListeners.remove(overrideSubscribeUrl);
registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
};
}
注意代码1处，有一个本地暴露，看下代码
private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker){
String key = getCacheKey(originInvoker);
ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
synchronized (bounds) {
exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
if (exporter == null) {
final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
exporter = new ExporterChangeableWrapper<T>((Exporter<T>)protocol.export(invokerDelegete), originInvoker);
bounds.put(key, exporter);
}
}
}
return (ExporterChangeableWrapper<T>) exporter;
}
getProviderUrl用于从远程暴露url中的export参数中获取本地暴露的url
private URL getProviderUrl(final Invoker<?> origininvoker){
String export = origininvoker.getUrl().getParameterAndDecoded(Constants.EXPORT_KEY);
if (export == null || export.length() == 0) {
throw new IllegalArgumentException("The registry export url is null! registry: " + origininvoker.getUrl());
}
URL providerUrl = URL.valueOf(export);
return providerUrl;
}
export参数对应的url才是需要实际本地暴露的，而作为export方法的远程暴露url只是为了注册提供者url到注册中心以及增加事件监听
同时注意一下bounds参数的校验是为了防止同一个invoker重复暴露，而ExporterChangeableWrapper封装是为了zookeeper中url发生改变时能修改Exporter
在完成本地暴露之后，会通过远程暴露url获取注册中心对象，然后把本地暴露url注册上去，同时也会给zookeeper中本地暴露url对应路径注册监听器，用于监听zookeeper上面的暴露url发生变化的时候，重新export（比如我们的控制台可以对参数进行调整）
最后把export返回
通过DubboProtocol暴露
DubboProtocol是使用远程协议的本地暴露，所以可以将暴露url注册到注册中心
看下它的export方法
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);
// modified by lishen
optimizeSerialization(url);
return exporter;
}
首先会把invoker转换为DubboExporter，放到exporterMap中
然后有一些stub的逻辑，这个以后单独再讲
接下来就是打开netty服务，用于监听服务引用者的请求，打开服务器逻辑在openServer中
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);
}
}
}
从String key = url.getAddress();以及ExchangeServer server = serverMap.get(key);可以看出来在一个应用中，netty服务器针对每种协议只会起一个，因为每种协议只能配置一个端口
而reset方法，会使用之后暴露url的参数，覆盖已经开启netty服务内的参数
那么我们的服务器是怎么处理接收的rpc请求并调用对应exporter调用呢，进入createServer方法我们可以看到会netty服务的开启时会绑定一个requestHandler
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 {
//1 绑定requestHandler
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;
}
这个requestHandler就是用来处理接收到的rpc调用请求的，看下它内部的逻辑
private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
if (message instanceof Invocation) {
Invocation inv = (Invocation) message;
//通过inv获取对应invoker
Invoker<?> invoker = getInvoker(channel, inv);
//如果是callback 需要处理高版本调用低版本的问题
if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))){
String methodsStr = invoker.getUrl().getParameters().get("methods");
boolean hasMethod = false;
if (methodsStr == null || methodsStr.indexOf(",") == -1){
hasMethod = inv.getMethodName().equals(methodsStr);
} else {
String[] methods = methodsStr.split(",");
for (String method : methods){
if (inv.getMethodName().equals(method)){
hasMethod = true;
break;
}
}
}
if (!hasMethod){
logger.warn(new IllegalStateException("The methodName "+inv.getMethodName()+" not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) +" ,invocation is :"+inv );
return null;
}
}
RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
//用invoker执行调用，返回结果
return invoker.invoke(inv);
}
throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
}
@Override
public void received(Channel channel, Object message) throws RemotingException {
if (message instanceof Invocation) {
reply((ExchangeChannel) channel, message);
} else {
super.received(channel, message);
}
}
@Override
public void connected(Channel channel) throws RemotingException {
invoke(channel, Constants.ON_CONNECT_KEY);
}
@Override
public void disconnected(Channel channel) throws RemotingException {
if(logger.isInfoEnabled()){
logger.info("disconected from "+ channel.getRemoteAddress() + ",url:" + channel.getUrl());
}
invoke(channel, Constants.ON_DISCONNECT_KEY);
}
private void invoke(Channel channel, String methodKey) {
Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
if (invocation != null) {
try {
received(channel, invocation);
} catch (Throwable t) {
logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
}
}
}
private Invocation createInvocation(Channel channel, URL url, String methodKey) {
String method = url.getParameter(methodKey);
if (method == null || method.length() == 0) {
return null;
}
RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
invocation.setAttachment(Constants.PATH_KEY, url.getPath());
invocation.setAttachment(Constants.GROUP_KEY, url.getParameter(Constants.GROUP_KEY));
invocation.setAttachment(Constants.INTERFACE_KEY, url.getParameter(Constants.INTERFACE_KEY));
invocation.setAttachment(Constants.VERSION_KEY, url.getParameter(Constants.VERSION_KEY));
if (url.getParameter(Constants.STUB_EVENT_KEY, false)){
invocation.setAttachment(Constants.STUB_EVENT_KEY, Boolean.TRUE.toString());
}
return invocation;
}
};
这边主要讲下replay方法，这个方法用来处理用户的rpc请求，请求的序列化dubbo封装的netty服务已经处理，所在在这个方法传入的message直接就是Invocation对象，在getInvoker中，通过invocation对象可以我们可以生成exporterMap的key，用来拿到对应的Exporter
Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException{
boolean isCallBackServiceInvoke = false;
boolean isStubServiceInvoke = false;
int port = channel.getLocalAddress().getPort();
String path = inv.getAttachments().get(Constants.PATH_KEY);
//如果是客户端的回调服务.
isStubServiceInvoke = Boolean.TRUE.toString().equals(inv.getAttachments().get(Constants.STUB_EVENT_KEY));
if (isStubServiceInvoke){
port = channel.getRemoteAddress().getPort();
}
//callback
isCallBackServiceInvoke = isClientSide(channel) && !isStubServiceInvoke;
if(isCallBackServiceInvoke){
path = inv.getAttachments().get(Constants.PATH_KEY)+"."+inv.getAttachments().get(Constants.CALLBACK_SERVICE_KEY);
inv.getAttachments().put(IS_CALLBACK_SERVICE_INVOKE, Boolean.TRUE.toString());
}
String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));
//2 通过invocation生成的key获取exporter
DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
if (exporter == null)
throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv);
//转换为invoker
return exporter.getInvoker();
}
拿到Exporter之后，转换为Invoker,直接调用invoke方法返回Result，之后返回给调用者的序列化等逻辑dubbo封装的netty服务也帮我们处理了，我们不用关注
这章主要讲解的是服务暴露，关于netty服务的实现不多分析(我也还没怎么看过)，理解这个requestHandler处理器即可，知道它会怎么处理rpc请求对应的Invocation即可，什么序列化，加密解密全都当作黑盒。
服务引用，下章见
最后
希望大家关注下我的公众号
image
