package org.apache.dubbo.rpc;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.common.extension.Adaptive;
import org.apache.dubbo.common.extension.SPI;
/**
* Protocol. (API/SPI, Singleton, ThreadSafe)
*/
@SPI("dubbo")
public interface Protocol {
/**
* 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;
}
public class HeaderExchanger implements Exchanger {
public static final String NAME = "header";
@Override
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
}
@Override
public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}
}
// all
// direct
// message 只有请求和响应消息派发到线程池
// execution 只有请求消息派发到线程池
// connection
ChannelHandler ch = ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)
我们进入AllChanelHandler看下里面的代码：
public class AllChannelHandler extends WrappedChannelHandler {
@Override
public void received(Channel channel, Object message) throws RemotingException {
// 获取线程池
ExecutorService executor = getExecutorService();
try {
executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
} catch (Throwable t) {
//TODO A temporary solution to the problem that the exception information can not be sent to the opposite end after the thread pool is full. Need a refactoring
//fix The thread pool is full, refuses to call, does not return, and causes the consumer to wait for time out
if(message instanceof Request && t instanceof RejectedExecutionException){
Request request = (Request)message;
if(request.isTwoWay()){
String msg = "Server side(" + url.getIp() + "," + url.getPort() + ") threadpool is exhausted ,detail msg:" + t.getMessage();
Response response = new Response(request.getId(), request.getVersion());
response.setStatus(Response.SERVER_THREADPOOL_EXHAUSTED_ERROR);
response.setErrorMessage(msg);
channel.send(response);
return;
}
}
throw new ExecutionException(message, channel, getClass() + " error when process received event .", t);
}
}
// 略
}
public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
// you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
// the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler
super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
}
public class FixedThreadPool implements ThreadPool {
@Override
public Executor getExecutor(URL url) {
String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME);
// 200
int threads = url.getParameter(THREADS_KEY, DEFAULT_THREADS);
int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES);
return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS,
queues == 0 ? new SynchronousQueue<Runnable>() :
(queues < 0 ? new LinkedBlockingQueue<Runnable>()
: new LinkedBlockingQueue<Runnable>(queues)),
new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url));
}
}
Apache Dubbo™ 是一款高性能Java RPC框架，这是一句来自官方的介绍。关于RPC框架的底层原理可以参见以前我写过的一篇文章：《写一个极简的RPC》。
关于dubbo的知识点官方文档已经讲得非常透彻了，做的好的做的不好的文档都有提及，本文将从源码层面理解dubbo的几个设计核心点。
基本格式：无处不在的URL
URL作为dubbo配置信息的统一格式，这是整个框架设计的基本原则，无论是注册中心，还是提供者、消费者都以URL形式存在。
上面这段URL就是一个典型的服务提供者，通过协议(dubbo)、服务地址(172.18.35.230:20880)和若干参数表达了Provider的信息。
上面这个URL是注册中心。
核心机制：Service Provider Interface
dubbo是一个采取了Microkernel + Plugin模式的框架，提供了很多组件的扩展点，做到这种模式我们就需要设计一套加载插件的方式，这个首要的机制就是SPI。
扩展点 @SPI
dubbo的SPI借鉴了JDK的SPI，但又有些许不同：
于是，我们在源码中可以发现大量的扩展点配置文件，我们以
org.apache.dubbo.rpc.Protocol接口为例，它的定义如下：Protocol的默认扩展有很多，在源码的不同maven项目(可以理解为不同jar包内部)下，可以看到有如下这些内容：注意到
Protocol的接口定义，所有SPI的服务接口都是用注解@SPI("dubbo")标识，dubbo参数值表示默认的扩展名称，对应配置文件的key。我们还注意到方法使用了注解
@Adaptive标识，这就是扩展自适应功能。扩展点自适应 @Adaptive
如何理解自适应这个概念呢？我们把它看成是运行时动态决定使用哪种扩展的方式。dubbo会在运行时对扩展服务动态生成一个代理类，这个类会在运行时动态决定使用哪一个扩展。如何动态决定扩展呢？Dubbo 使用 URL 对象（包含了Key-Value）传递配置信息，扩展点方法调用会有URL参数（或是参数有URL成员）。
我们先来看看如何获得自适应类的对象，注意它是一个单例：
现在的重点就是看下如何生成这样的代理类，它是通过运行时生成源码文件，然后调用编译器进行编译后生成，这里贴上生成后的代码：
从代码中读出以下关键信息：
我们看下生成代理类的源码：
可以看到编译器也是一个扩展点，需要使用扩展的地方都可以通过扩展机制加载。关于如何生成代理类，如何从配置文件中加载扩展类，以及如何实现扩展点AOP和IOC功能的核心源码都在ExtensionLoader类中，这里就不赘述。
领域模型
整个源码中有很多领域封装，有的领域在框架分层设计的某一个层中，有的领域贯穿整个框架。
接下来我们将关注几个核心领域模型，它们分别是Protocol、Invoker、Invocation。
Protocol服务域
Protocol是服务导出和服务引入的核心领域模型，再看一次Protocol的方法：
这两个方法在整个dubbo的源码中是服务的核心抽象，export方法导出一个服务，refer方法引用一个服务。接口的实现关系如下：
其中RegistryProtocol实现了注册中心相关功能，具体的DubboProtocol则实现了启动netty server和client端相关功能，关于服务导出和引入的流程请参见官网。
Invoker调用实体域
仔细观察Protocol的方法发现，无论是导出或是引入服务都有一个很重要的领域：Invoker，它代表一个可执行体，可向它发起调用。
invoke方法代表执行一次调用返回结果。服务导出时通过代理工厂生成实例，生成Invoker的源码如下：
服务引入时生成Invoker经过了一些封装，我们先来看下Protocol的refer方法默认实现：
AsyncToSyncInvoker是一个异步转同步的Invoker，protocolBindingRefer方法在DubboProtocol的源码如下：
可以看到核心的调用逻辑封装在DubboInvoker中，它会发起调用请求，并且返回一个Result结果，关于Result将会在下文的线程派发模型中进一步介绍。
在整个服务引用方调用服务链路中，dubbo还对集群容错做了一些处理，在实际引用中会对AsyncToSyncInvoker进一步的做封装，关于集群容错的默认实现类是：FailOverClusterInvoker，代表故障切换，还有一些可选容错模式：
FailOverClusterInvoker内部通过Directory保存了Invoker的列表，Directory能感知到注册中心Invoker的变化，我们来看下源码：
dubbo负载均衡的原理就从这里开始，进一步的知识可以继续跟进doInvoker的源码。
Invocation会话域
在Invoker实体域的方法invoke拥有一个参数，类型为Invocation，它持有调用过程中的变量，比如方法名，参数等，默认实现是RpcInvocation，底层网络通信都会对Invocation进行编码和解码。
扩展模块
本小节介绍下一些核心的扩展模块，比如网络通信模块、注册中心模块，这些模块在dubbo框架中组合了一些开源的实现方案。
Netty
Netty俨然已经成为现在Java网络通信的首选组件，它是一个高性能的异步事件驱动网络框架。
Dubbo默认使用Netty为底层通信，核心功能是建立网络连接，编解码，发送消息和接受消息。Netty的源码实现为
NettyServer和NettyClient，分别对应服务端和消费端。我们看下NettyServer启动的源码：熟悉Netty的对这段代码应该很熟悉，这是典型的Netty服务端代码：
为了做到扩展，NettyServer和NettyClient都被封装到HeaderExchangeServer和HeaderExchangeClient类，我们可以通过扩展自适应机制设置底层通信框架为netty或者mina，Exchange层的实现类为HeaderExchanger，提供了创建服务端和客户端连接服务端方法：
Zookeeper
在RPC或者微服务中服务的管理变得尤为关键，这个时候就需要一个统一服务注册和管理的中心，Zookeeper作为一个高可用的分布式协调服务，默认成为dubbo的注册中心。
dubbo中注册中心的实现为ZookeeperRegistry，提供了注册和订阅变更的功能，它继承了
FailbackRegistry类，这是一个失败自动恢复的注册中心实现，恢复的源码如下：这里会自动恢复注册和订阅，ZookeeperRegistry在初始化时会监听重连事件，触发恢复机制。
注册中心在dubbo框架的位置是举足轻重的，如果它出问题就会产生单点故障，所以必须保证注册中心的高可用。dubbo在注册中心的源码中，提供了注册中心本地缓存机制进行些许的容错，在注册中心抽象实现
AbstractRegistry的构造器中加载了本地缓存：这里尤其要注意的是，容错的方案是针对订阅的容错，不是针对注册的容错，关于这部分的设计问题个人还有不少疑问，有待进一步考察。
序列化
dubbo默认采用hessian2的序列化方案，这里不深入介绍，贴上序列化接口的源码：
线程派发模型
上面这张图表达的是提供方服务端的线程派发，在消费方客户端也有着类似的线程派发模型。
Dispatcher派发 和 线程池
Dispatcher根据配置来决定线程派发方式，默认的配置为all，所有消息都派发到线程池，Dispatcher的核心功能是创建合适的ChannelHandler，当配置为all时，ChannelHandler为AllChannelHandler。
我们进入AllChanelHandler看下里面的代码：
这段可以看到获取线程池执行任务的代码，那么问题来了：这是一个什么样的线程池，我该如何优化它？
这个线程池是在NettyServer和NettyClient启动时创建ChannelHandler的同时初始化的，我们先来看下NettyServer中创建ChannelHandler的代码：
ChannelHandlers.wrap方法用来创建合适的ChannelHandler。我们以AllChannelHandler为例，在初始化时会调用父类
WrappedChannelHandler的构造函数：看到这里已经很明朗了，线程池是
ThreadPool的一个扩展，它拥有一个返回线程池的方法Executor getExecutor(URL url);。dubbo提供方服务端默认的线程池采用的是fixed类型，默认线程个数为200，我们看下构造线程池的代码：同理，很重要的一点是消费客户端方也拥有类似的线程模型，通过线程池执行响应处理等任务，那么它的线程池是什么样的呢？答案默认是cached，原理是在构造NettyClient的时候会默认给URL添加
threadpool=cached参数，这样在自适应扩展时就会获取CachedThreadPool实例，NettyClient构造代码如下：DefaultFuture
线程派发模型中有个很重要的问题需要解决，就是线程池处理的响应结果如何传递给用户线程并且唤醒用户线程？带着这个问题我们先来看下DubboInvoker是如何调用远程服务的：
AsyncRpcResult继承了CompletableFuture<Result>类，从而实现异步编程。responseFuture也是一个CompletableFuture<Result>类型的对象，默认实现是DefaultFuture，所以上面的问题就变成了：当有响应结果返回时，如何唤醒合适的DefaultFuture？答案就在
DefaultFuture源码中，这是一个比较重的对象，存储了所有的DefaultFuture实例，当接受到响应时，会调用静态方法received：FUTURES存储了所有实例，是一个Map结构，key是一个long类型的编号，而这个编号就是关键：
所以唤醒用户线程是通过一个编号来关联的。
写在最后
成年人的世界里设计总没有对错，dubbo有它独特的整体设计，虽然某些细节还不能尽善尽美。理解dubbo内部的工作细节，才能对一个框架的使用保持最合适的态度。