Dubbo源码分析系列-深入RPC协议扩展-小白筆記-匯編語言學習筆記

Dubbo源码分析系列-深入RPC协议扩展

2023-09-15 阅读 19 评论 0

摘要：导语在之前的博客里面提到了关于扩展机制以及SPI的原理，这篇博客主要来讨论一下关于协议的扩展问题，在系统与系统之间通信就需要两个系统之间遵循相同的协议。而现在被熟知的常用的协议有TCP/IP协议、HTTP协议等等。在Dubbo中也提供了属于Dubbo的专属协议

导语
在之前的博客里面提到了关于扩展机制以及SPI的原理，这篇博客主要来讨论一下关于协议的扩展问题，在系统与系统之间通信就需要两个系统之间遵循相同的协议。而现在被熟知的常用的协议有TCP/IP协议、HTTP协议等等。在Dubbo中也提供了属于Dubbo的专属协议dubbo协议。下面就来详细的了解一下关于Dubbo协议。

协议扩展简介

首先需要知道的Dubbo是基于RPC进行远程调用的框架。它封装了很多的关于RPC的远程调用细节。这里先来看看Dubbo对于协议的封装接口org.apache.dubbo.rpc.Protocol，这个接口是Apache公司重新进行定义的。

@SPI("dubbo")
public interface Protocol {/*** Get default port when user doesn't config the port.** @return default port*/int getDefaultPort();/*** Export service for remote invocation: <br>* 1. Protocol should record request source address after receive a request:* RpcContext.getContext().setRemoteAddress();<br>* 2. export() must be idempotent, that is, there's no difference between invoking once and invoking twice when* export the same URL<br>* 3. Invoker instance is passed in by the framework, protocol needs not to care <br>** @param <T>     Service type* @param invoker Service invoker* @return exporter reference for exported service, useful for unexport the service later* @throws RpcException thrown when error occurs during export the service, for example: port is occupied*/@Adaptive<T> Exporter<T> export(Invoker<T> invoker) throws RpcException;/*** Refer a remote service: <br>* 1. When user calls `invoke()` method of `Invoker` object which's returned from `refer()` call, the protocol* needs to correspondingly execute `invoke()` method of `Invoker` object <br>* 2. It's protocol's responsibility to implement `Invoker` which's returned from `refer()`. Generally speaking,* protocol sends remote request in the `Invoker` implementation. <br>* 3. When there's check=false set in URL, the implementation must not throw exception but try to recover when* connection fails.** @param <T>  Service type* @param type Service class* @param url  URL address for the remote service* @return invoker service's local proxy* @throws RpcException when there's any error while connecting to the service provider*/@Adaptive<T> Invoker<T> refer(Class<T> type, URL url) throws RpcException;/*** Destroy protocol: <br>* 1. Cancel all services this protocol exports and refers <br>* 2. Release all occupied resources, for example: connection, port, etc. <br>* 3. Protocol can continue to export and refer new service even after it's destroyed.*/void destroy();}

在Java开发中对于接口的作用就是用来定义一些被其他类继承的规则。当实现类继承了这个接口之后，就必须实现这个接口中的所有方法。这里可以看到，这个接口提供了如下的四个方法。
在这里插入图片描述

当用户调用refer()返回的Invoker对象的invoke()方法的时候，需要先执行远端export()方法传入的Invoker对象的Invoke()方法。
refer()返回的Invoker是由协议实现，协议通常需要在Invoker中进行远程请求，export()传入的Invoker由框架实现并进行传递。也就是说服务提供方在容器启动的时候就进行服务的暴露，而服务调用方需要通过协议进行Invoker的调用。

qq协议源码。在这里协议并不会直接访问需要调用的对象，而是访问对象的代理，也就是说在协议调用的过程中会把Invoker转换为一个对应的接口。这里的协议并不一定是基于TCP的网络通信协议，还可以是其他的文件协议或者是进程之间的通信方式等等。

扩展接口

在Dubbo中其实提供了很多的扩展的接口除了上面提到的org.apache.dubbo.rpc.Protocol接口以外还提供了如下的一些接口

org.apache.dubbo.rpc.Exporter
org.apache.dubbo.rpc.Invoker

org.apache.dubbo.rpc.Invoker


/*** Invoker. (API/SPI, Prototype, ThreadSafe)** @see org.apache.dubbo.rpc.Protocol#refer(Class, org.apache.dubbo.common.URL)* @see org.apache.dubbo.rpc.InvokerListener* @see org.apache.dubbo.rpc.protocol.AbstractInvoker*/
public interface Invoker<T> extends Node {/*** get service interface.** @return service interface.*/Class<T> getInterface();/*** invoke.** @param invocation* @return result* @throws RpcException*/Result invoke(Invocation invocation) throws RpcException;}

org.apache.dubbo.rpc.Exporter

/*** Exporter. (API/SPI, Prototype, ThreadSafe)** @see org.apache.dubbo.rpc.Protocol#export(Invoker)* @see org.apache.dubbo.rpc.ExporterListener* @see org.apache.dubbo.rpc.protocol.AbstractExporter*/
public interface Exporter<T> {/*** get invoker.** @return invoker*/Invoker<T> getInvoker();/*** unexport.* <p>* <code>* getInvoker().destroy();* </code>*/void unexport();}

购买源代码协议。对于这些提供扩展的接口来说毫无疑问是需要继承这些接口的子类实现其中的方法。而这里会看到在Invoker接口上继承了一个新Node接口。定义如下,那么为什么会提供这样的一个接口呢。首先可以看到这个Node接口中提供了三个方法。分别是获取到url、判断是否可用以及进行销毁操作。在上面我们提到过一个概念就是，Invoker是基于协议的，而对于一个协议来说最主要的就是找到调用方在什么地方以及调用方是否可用。这里定义这个Node类的作用就是为了标记这个服务的位置以及健康状况。

/*** Node. (API/SPI, Prototype, ThreadSafe)*/
public interface Node {/*** get url.** @return url.*/URL getUrl();/*** is available.** @return available.*/boolean isAvailable();/*** destroy.*/void destroy();}

在上面我们可以看到Dubbo对于协议的扩展，那么Dubbo是怎么提供对于协议的配置呢？在Dubbo给出的示例中。在Dubbo源码的demo模块中提供的基于XML配置文件的形式。给出了服务提供方和服务消费方的配置
服务提供方

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"xmlns="http://www.springframework.org/schema/beans"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsdhttp://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">//服务应用名称<dubbo:application name="demo-provider"/>//注册中心的地址<dubbo:registry address="zookeeper://127.0.0.1:2181"/>//支持协议<dubbo:protocol name="dubbo"/>//服务提供方向Spring容器中注入服务<bean id="demoService" class="org.apache.dubbo.demo.provider.DemoServiceImpl"/>//暴露服务接口<dubbo:service interface="org.apache.dubbo.demo.DemoService" ref="demoService"/></beans>

服务消费方

<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"xmlns="http://www.springframework.org/schema/beans"xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-4.3.xsdhttp://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">//服务消费方应用名称<dubbo:application name="demo-consumer"/>//服务调用注册中心<dubbo:registry address="zookeeper://127.0.0.1:2181"/>//服务提供方<dubbo:reference id="demoService" check="false" interface="org.apache.dubbo.demo.DemoService"/></beans>

Dubbo已知协议扩展类

在这里插入图片描述
如图所示，在Dubbo RPC模块中提供了很多的协议协议扩展模块。下面就来看看Dubbo提供的Dubbo模块中的扩展协议。org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol在这个路径下。

在这个src路径下面提供了实现类，在resource路径下面提供了如下图所示的RPC配置。这里可以看到在Dubbo中对于协议、Filter等都做了扩展。从上面的分析中我们知道在Protocol中最为主要的两个方法就是export()服务暴露方法和refer()服务调用方法。这里我们从org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol类中开始分析。
在这里插入图片描述
在DubboProtocol类中找到export方法，如下可以看到提供的都是泛型调用。从上面我们知道Invoker其实是真实对象的代理而Exporter则是为了获取到Invoker对象以及其代理的对象。也就是说在实际协议传输的时候操作的其实是Invoker并不会破坏实际对象的封装。

安卓QQ协议源码，对于DubboProtocol的扩展类
在这里插入图片描述
对于export()方法的实现

 @Overridepublic <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 dispatching eventBoolean 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);}}openServer(url);optimizeSerialization(url);return exporter;}

对于refer()方法的实现

 @Overridepublic <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));}

@Override
public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {optimizeSerialization(url);// create rpc invoker.DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);invokers.add(invoker);return invoker;
}

refer()是其抽象父类org.apache.dubbo.rpc.protocol.AbstractProtocol中被实现的，在调用过程中被子类DubboProtocol的protocolBindingRefer()方法实现。并且通过下面的方法对于Invoker进行了缓存操作。这样做的目的是可以根据子类的实现方式来控制抽象父类的方式。也有些类似于子类实现该方法实现子类专属功能。实现了真实调用与实际调用的分离，能更好的通过实际调用来扩展调用方法的实现功能。例如上面这个做法实际上就是通过子类的方法对于Invoker在子类中进行和缓存操作，而这样做的目的就是在其他方法中可以更为高效的进行Invoker的调用。

//TODO SoftReference
protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>();

对于DubboExporter的扩展类
在这里插入图片描述

public class DubboExporter<T> extends AbstractExporter<T> {private final String key;private final Map<String, Exporter<?>> exporterMap;public DubboExporter(Invoker<T> invoker, String key, Map<String, Exporter<?>> exporterMap) {super(invoker);this.key = key;this.exporterMap = exporterMap;}@Overridepublic void unexport() {super.unexport();exporterMap.remove(key);}}

public abstract class AbstractExporter<T> implements Exporter<T> {protected final Logger logger = LoggerFactory.getLogger(getClass());private final Invoker<T> invoker;private volatile boolean unexported = false;public AbstractExporter(Invoker<T> invoker) {if (invoker == null) {throw new IllegalStateException("service invoker == null");}if (invoker.getInterface() == null) {throw new IllegalStateException("service type == null");}if (invoker.getUrl() == null) {throw new IllegalStateException("service url == null");}this.invoker = invoker;}@Overridepublic Invoker<T> getInvoker() {return invoker;}@Overridepublic void unexport() {if (unexported) {return;}unexported = true;getInvoker().destroy();}@Overridepublic String toString() {return getInvoker().toString();}}