Dubbo源码解析（二十二）远程调用——Protocol
目标：介绍远程调用中协议的设计和实现，介绍dubbo-rpc-api中的各种protocol包的源码，是重点内容。
前言
在远程调用中协议是非常重要的一层，看下面这张图：
该层是在信息交换层之上，分为了并且夹杂在服务暴露和服务引用中间，为了有一个约定的方式进行调用。
dubbo支持不同协议的扩展，比如http、thrift等等，具体的可以参照官方文档。本文讲解的源码大部分是对于公共方法的实现，而具体的服务暴露和服务引用会在各个协议实现中讲到。
下面是该包下面的类图：
源码分析
（一）AbstractProtocol
该类是协议的抽象类，实现了Protocol接口，其中实现了一些公共的方法，抽象方法在它的子类AbstractProxyProtocol中定义。
1.属性
/** * 服务暴露者集合 */ protected final Map<String, Exporter<?>> exporterMap = new ConcurrentHashMap<String, Exporter<?>>(); /** * 服务引用者集合 */ //TODO SOFEREFENCE protected final Set<Invoker<?>> invokers = new ConcurrentHashSet<Invoker<?>>(); 复制代码
2.serviceKey
protected static String serviceKey(URL url) {
// 获得绑定的端口号
int port = url.getParameter(Constants.BIND_PORT_KEY, url.getPort());
return serviceKey(port, url.getPath(), url.getParameter(Constants.VERSION_KEY),
url.getParameter(Constants.GROUP_KEY));
}
protected static String serviceKey(int port, String serviceName, String serviceVersion, String serviceGroup) {
return ProtocolUtils.serviceKey(port, serviceName, serviceVersion, serviceGroup);
}
复制代码
该方法是为了得到服务key group+"/"+serviceName+":"+serviceVersion+":"+port
3.destroy
@Override
public void destroy() {
// 遍历服务引用实体
for (Invoker<?> invoker : invokers) {
if (invoker != null) {
// 从集合中移除
invokers.remove(invoker);
try {
if (logger.isInfoEnabled()) {
logger.info("Destroy reference: " + invoker.getUrl());
}
// 销毁
invoker.destroy();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
// 遍历服务暴露者
for (String key : new ArrayList<String>(exporterMap.keySet())) {
// 从集合中移除
Exporter<?> exporter = exporterMap.remove(key);
if (exporter != null) {
try {
if (logger.isInfoEnabled()) {
logger.info("Unexport service: " + exporter.getInvoker().getUrl());
}
// 取消暴露
exporter.unexport();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
}
复制代码
该方法是对invoker和exporter的销毁。
（二）AbstractProxyProtocol
该类继承了AbstractProtocol类，其中利用了代理工厂对AbstractProtocol中的两个集合进行了填充，并且对异常做了处理。
1.属性
/** * rpc的异常类集合 */ private final List<Class<?>> rpcExceptions = new CopyOnWriteArrayList<Class<?>>(); /** * 代理工厂 */ private ProxyFactory proxyFactory; 复制代码
2.export
@Override
@SuppressWarnings("unchecked")
public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
// 获得uri
final String uri = serviceKey(invoker.getUrl());
// 获得服务暴露者
Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
if (exporter != null) {
return exporter;
}
// 新建一个线程
final Runnable runnable = doExport(proxyFactory.getProxy(invoker, true), invoker.getInterface(), invoker.getUrl());
exporter = new AbstractExporter<T>(invoker) {
/**
* 取消暴露
*/
@Override
public void unexport() {
super.unexport();
// 移除该key对应的服务暴露者
exporterMap.remove(uri);
if (runnable != null) {
try {
// 启动线程
runnable.run();
} catch (Throwable t) {
logger.warn(t.getMessage(), t);
}
}
}
};
// 加入集合
exporterMap.put(uri, exporter);
return exporter;
}
复制代码
其中分为两个步骤，创建一个exporter，放入到集合汇中。在创建exporter时对unexport方法进行了重写。
3.refer
@Override
public <T> Invoker<T> refer(final Class<T> type, final URL url) throws RpcException {
// 通过代理获得实体域
final Invoker<T> target = proxyFactory.getInvoker(doRefer(type, url), type, url);
Invoker<T> invoker = new AbstractInvoker<T>(type, url) {
@Override
protected Result doInvoke(Invocation invocation) throws Throwable {
try {
// 获得调用结果
Result result = target.invoke(invocation);
Throwable e = result.getException();
// 如果抛出异常，则抛出相应异常
if (e != null) {
for (Class<?> rpcException : rpcExceptions) {
if (rpcException.isAssignableFrom(e.getClass())) {
throw getRpcException(type, url, invocation, e);
}
}
}
return result;
} catch (RpcException e) {
// 抛出未知异常
if (e.getCode() == RpcException.UNKNOWN_EXCEPTION) {
e.setCode(getErrorCode(e.getCause()));
}
throw e;
} catch (Throwable e) {
throw getRpcException(type, url, invocation, e);
}
}
};
// 加入集合
invokers.add(invoker);
return invoker;
}
复制代码
该方法是服务引用，先从代理工厂中获得Invoker对象target，然后创建了真实的invoker在重写方法中调用代理的方法，最后加入到集合。
protected abstract <T> Runnable doExport(T impl, Class<T> type, URL url) throws RpcException; protected abstract <T> T doRefer(Class<T> type, URL url) throws RpcException; 复制代码
可以看到其中抽象了服务引用和暴露的方法，让各类协议各自实现。
(三)AbstractInvoker
该类是invoker的抽象方法，因为协议被夹在服务引用和服务暴露中间，无论什么协议都有一些通用的Invoker和exporter的方法实现，而该类就是实现了Invoker的公共方法，而把doInvoke抽象出来，让子类只关注这个方法。
1.属性
/** * 服务类型 */ private final Class<T> type; /** * url对象 */ private final URL url; /** * 附加值 */ private final Map<String, String> attachment; /** * 是否可用 */ private volatile boolean available = true; /** * 是否销毁 */ private AtomicBoolean destroyed = new AtomicBoolean(false); 复制代码
2.convertAttachment
private static Map<String, String> convertAttachment(URL url, String[] keys) {
if (keys == null || keys.length == 0) {
return null;
}
Map<String, String> attachment = new HashMap<String, String>();
// 遍历key，把值放入附加值集合中
for (String key : keys) {
String value = url.getParameter(key);
if (value != null && value.length() > 0) {
attachment.put(key, value);
}
}
return attachment;
}
复制代码
该方法是转化为附加值，把url中的值转化为服务调用invoker的附加值。
3.invoke
@Override
public Result invoke(Invocation inv) throws RpcException {
// if invoker is destroyed due to address refresh from registry, let's allow the current invoke to proceed
// 如果服务引用销毁，则打印告警日志，但是通过
if (destroyed.get()) {
logger.warn("Invoker for service " + this + " on consumer " + NetUtils.getLocalHost() + " is destroyed, "
+ ", dubbo version is " + Version.getVersion() + ", this invoker should not be used any longer");
}
RpcInvocation invocation = (RpcInvocation) inv;
// 会话域中加入该调用链
invocation.setInvoker(this);
// 把附加值放入会话域
if (attachment != null && attachment.size() > 0) {
invocation.addAttachmentsIfAbsent(attachment);
}
// 把上下文的附加值放入会话域
Map<String, String> contextAttachments = RpcContext.getContext().getAttachments();
if (contextAttachments != null && contextAttachments.size() != 0) {
/**
* invocation.addAttachmentsIfAbsent(context){@link RpcInvocation#addAttachmentsIfAbsent(Map)}should not be used here,
* because the {@link RpcContext#setAttachment(String, String)} is passed in the Filter when the call is triggered
* by the built-in retry mechanism of the Dubbo. The attachment to update RpcContext will no longer work, which is
* a mistake in most cases (for example, through Filter to RpcContext output traceId and spanId and other information).
*/
invocation.addAttachments(contextAttachments);
}
// 如果开启的是异步调用，则把该设置也放入附加值
if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)) {
invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
}
// 加入编号
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
// 执行调用链
return doInvoke(invocation);
} catch (InvocationTargetException e) { // biz exception
Throwable te = e.getTargetException();
if (te == null) {
return new RpcResult(e);
} else {
if (te instanceof RpcException) {
((RpcException) te).setCode(RpcException.BIZ_EXCEPTION);
}
return new RpcResult(te);
}
} catch (RpcException e) {
if (e.isBiz()) {
return new RpcResult(e);
} else {
throw e;
}
} catch (Throwable e) {
return new RpcResult(e);
}
}
复制代码
该方法做了一些公共的操作，比如服务引用销毁的检测，加入附加值，加入调用链实体域到会话域中等。然后执行了doInvoke抽象方法。各协议自己去实现。
（四）AbstractExporter
该类和AbstractInvoker类似，也是在服务暴露中实现了一些公共方法。
1.属性
/** * 实体域 */ private final Invoker<T> invoker; /** * 是否取消暴露服务 */ private volatile boolean unexported = false; 复制代码
2.unexport
@Override
public void unexport() {
// 如果已经消取消暴露，则之间返回
if (unexported) {
return;
}
// 设置为true
unexported = true;
// 销毁该实体域
getInvoker().destroy();
}
复制代码
（五）InvokerWrapper
该类是Invoker的包装类，其中用到类装饰模式，不过并没有实现实际的功能增强。
public class InvokerWrapper<T> implements Invoker<T> {
/**
* invoker对象
*/
private final Invoker<T> invoker;
private final URL url;
public InvokerWrapper(Invoker<T> invoker, URL url) {
this.invoker = invoker;
this.url = url;
}
@Override
public Class<T> getInterface() {
return invoker.getInterface();
}
@Override
public URL getUrl() {
return url;
}
@Override
public boolean isAvailable() {
return invoker.isAvailable();
}
@Override
public Result invoke(Invocation invocation) throws RpcException {
return invoker.invoke(invocation);
}
@Override
public void destroy() {
invoker.destroy();
}
}
复制代码
（六）ProtocolFilterWrapper
该类实现了Protocol接口，其中也用到了装饰模式，是对Protocol的装饰，是在服务引用和暴露的方法上加上了过滤器功能。
1.buildInvokerChain
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()) {
// 从最后一个过滤器开始循环，创建一个带有过滤器链的invoker对象
for (int i = filters.size() - 1; i >= 0; i--) {
final Filter filter = filters.get(i);
// 记录last的invoker
final Invoker<T> next = last;
// 新建last
last = new Invoker<T>() {
@Override
public Class<T> getInterface() {
return invoker.getInterface();
}
@Override
public URL getUrl() {
return invoker.getUrl();
}
@Override
public boolean isAvailable() {
return invoker.isAvailable();
}
/**
* 关键在这里，调用下一个filter代表的invoker，把每一个过滤器串起来
* @param invocation
* @return
* @throws RpcException
*/
@Override
public Result invoke(Invocation invocation) throws RpcException {
return filter.invoke(next, invocation);
}
@Override
public void destroy() {
invoker.destroy();
}
@Override
public String toString() {
return invoker.toString();
}
};
}
}
return last;
}
复制代码
该方法就是创建带 Filter 链的 Invoker 对象。倒序的把每一个过滤器串连起来，形成一个invoker。
2.export
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
// 如果是注册中心，则直接暴露服务
if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) {
return protocol.export(invoker);
}
// 服务提供侧暴露服务
return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
}
复制代码
该方法是在服务暴露上做了过滤器链的增强，也就是加上了过滤器。
3.refer
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 如果是注册中心，则直接引用
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
return protocol.refer(type, url);
}
// 消费者侧引用服务
return buildInvokerChain(protocol.refer(type, url), Constants.REFERENCE_FILTER_KEY, Constants.CONSUMER);
}
复制代码
该方法是在服务引用上做了过滤器链的增强，也就是加上了过滤器。
（七）ProtocolListenerWrapper
该类也实现了Protocol，也是装饰了Protocol接口，但是它是在服务引用和暴露过程中加上了监听器的功能。
1.export
@Override
public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
// 如果是注册中心，则暴露该invoker
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)));
}
复制代码
该方法是在服务暴露上做了监听器功能的增强，也就是加上了监听器。
2.refer
@Override
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
// 如果是注册中心。则直接引用服务
if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
return protocol.refer(type, url);
}
// 创建引用服务监听器包装类对象
return new ListenerInvokerWrapper<T>(protocol.refer(type, url),
Collections.unmodifiableList(
ExtensionLoader.getExtensionLoader(InvokerListener.class)
.getActivateExtension(url, Constants.INVOKER_LISTENER_KEY)));
}
复制代码
该方法是在服务引用上做了监听器功能的增强，也就是加上了监听器。
正文到此结束
- 本文标签： 端口 IDE 实例 provider 监听器 id 类图 源码 UI 遍历 dubbo trigger Collection retry 协议 map 解析 final ACE IO Proxy update cat App volatile http ORM equals 线程 CTO list src constant ArrayList tar CEO value build Service ConcurrentHashMap 代码 key example Atom HashMap https API Collections consumer message 注册中心 HashSet
- 版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
- 本文海报： 生成海报一 生成海报二
热门推荐
相关文章
近期评论
-
666
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
