文章目录
- 引言
- 服务的交互
- 服务降级
- 集群容错
- 服务调用
- 服务端接收请求
- 总结
引言
经过之前文章的铺垫，现在可以来分析服务的交互调用过程了。
服务的交互
服务降级
从名字上看我们不难理解MockClusterInvoker相当于是对普通Invoker（如DubboInvoker）的装饰增强，提供集群容错相关的功能，因此最终还是会进入到DubboInvoker，所以这里就以MockClusterInvoker.invoke方法开始分析：
代码语言：javascript复制public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
// 获取mock配置
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
if (value.length() == 0 || value.equalsIgnoreCase("false")){
// 未配置mock
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
// 强制执行mock逻辑，不发起远程调用
result = doMockInvoke(invocation, null);
} else {
// 首先调用远程服务，失败后执行mock逻辑，不抛出异常
try {
result = this.invoker.invoke(invocation);
}catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
result = doMockInvoke(invocation, e);
}
}
}
return result;
}Mock服务就是提供一个替代的服务接口，也就是服务降级，从这里看出有三种配置方式：直接调用服务、直接调用mock服务以及远程服务调用失败后调用mock服务。服务降级这里暂不分析，咱们接着看远程服务调用this.invoker.invoke，这个invoker对象上文分析过了，是FailoverClusterInvoker对象（即调用失败后重试其它远程服务），不过该类中没有invoke方法，所以是调用父类的AbstractClusterInvoker.invoke：
代码语言：javascript复制public Result invoke(final Invocation invocation) throws RpcException {
checkWheatherDestoried();
LoadBalance loadbalance;
List<Invoker<T>> invokers = list(invocation);
// 获取负载均衡配置，默认使用的是随机算法
if (invokers != null && invokers.size() > 0) {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(invokers.get(0).getUrl()
.getMethodParameter(invocation.getMethodName(),Constants.LOADBALANCE_KEY, Constants.DEFAULT_LOADBALANCE));
} else {
loadbalance = ExtensionLoader.getExtensionLoader(LoadBalance.class).getExtension(Constants.DEFAULT_LOADBALANCE);
}
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
// 具体的容错机制有子类实现
return doInvoke(invocation, invokers, loadbalance);
}这里首先会获取配置的负载均衡算法，默认是使用RandomLoadBalance，不过这个并不是简单的随机，而是加入了权重值，感兴趣的可以去看看其doSelect方法的实现。
集群容错
拿到负载均衡器后，会传入到集群容错的的具体实现类中进行选择。Dubbo默认使用的是FailoverClusterInvoker失败切换容错机制，所以这里以此为例，其它的容错机制读者可自行分析。
代码语言：javascript复制public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
// 检查invoker列表是否变更
checkInvokers(copyinvokers, invocation);
// 获取需要重试的次数，即invoker个数
int len = getUrl().getMethodParameter(invocation.getMethodName(), Constants.RETRIES_KEY, Constants.DEFAULT_RETRIES) 1;
if (len <= 0) {
len = 1;
}
// retry loop.
RpcException le = null; // last exception.
// 已经调用过的
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size());
Set<String> providers = new HashSet<String>(len);
for (int i = 0; i < len; i ) {
//重试时，进行重新选择，避免重试时invoker列表已发生变化.
//注意：如果列表发生了变化，那么invoked判断会失效，因为invoker示例已经改变
if (i > 0) {
checkWheatherDestoried();
copyinvokers = list(invocation);
//重新检查一下
checkInvokers(copyinvokers, invocation);
}
// 负载均衡筛选出的invoker
Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
invoked.add(invoker);
RpcContext.getContext().setInvokers((List)invoked);
try {
// 调用对应的invoker，执行成功返回对应的结果，否则出现异常则重试其它服务
Result result = invoker.invoke(invocation);
return result;
} catch (RpcException e) {
if (e.isBiz()) { // biz exception.
throw e;
}
le = e;
} catch (Throwable e) {
le = new RpcException(e.getMessage(), e);
} finally {
providers.add(invoker.getUrl().getAddress());
}
}
// 重试失败则抛出异常
throw new RpcException(le != null ? le.getCode() : 0, "Failed to invoke the method "
invocation.getMethodName() " in the service " getInterface().getName()
". Tried " len " times of the providers " providers
" (" providers.size() "/" copyinvokers.size()
") from the registry " directory.getUrl().getAddress()
" on the consumer " NetUtils.getLocalHost() " using the dubbo version "
Version.getVersion() ". Last error is: "
(le != null ? le.getMessage() : ""), le != null && le.getCause() != null ? le.getCause() : le);
}该容错机制的逻辑就是根据重试次数进行循环调用，每次循环都会通过select挑选出一个代理对象进行远程调用，若成功则直接返回结果，否则直到达到重试次数抛出异常。select方法中还有一个细节需要注意，我们来看一下：
代码语言：javascript复制protected Invoker<T> select(LoadBalance loadbalance, Invocation invocation, List<Invoker<T>> invokers, List<Invoker<T>> selected) throws RpcException {
if (invokers == null || invokers.size() == 0)
return null;
String methodName = invocation == null ? "" : invocation.getMethodName();
// 获取sticky粘滞连接配置，该配置表示是否始终使用同一个服务，除非该服务挂掉
boolean sticky = invokers.get(0).getUrl().getMethodParameter(methodName,Constants.CLUSTER_STICKY_KEY, Constants.DEFAULT_CLUSTER_STICKY) ;
{
// invoker列表中已经没有该粘滞连接对象，表示该服务已挂掉
if ( stickyInvoker != null && !invokers.contains(stickyInvoker) ){
stickyInvoker = null;
}
// 使用粘滞连接需要保证该连接不在invoked列表中，该列表中的服务表示存活且已经调用过的服务，
// 但由于某种原因未提供正常的服务，粘滞连接也不再去调用该服务
if (sticky && stickyInvoker != null && (selected == null || !selected.contains(stickyInvoker))){
// 确保粘滞连接是可用的
if (availablecheck && stickyInvoker.isAvailable()){
return stickyInvoker;
}
}
}
// 使用负载均衡算法挑选出invoker
Invoker<T> invoker = doselect(loadbalance, invocation, invokers, selected);
if (sticky){
stickyInvoker = invoker;
}
return invoker;
}该方法的逻辑是第一次调用先通过负载均衡算法获取服务，若配置了粘滞连接且该服务正常，则在后续调用中一直使用该服务，直到该服务不可用时再使用负载均衡算法获取服务。 以上就是服务调用前需要执行的保护机制，下面我们就来看看具体的服务调用实现。
服务调用
Result result = invoker.invoke是执行服务调用的逻辑，这里的invoker不难理解由具体的协议决定，这里就是DubboInvoker，同样的，DubboInvoker没有该方法的实现，是继承AbstractInvoker的，可以看到Dubbo中大量使用模板方法模式抽离公共的逻辑，因此具体的实现还是在DubboInvoker的doInvoke方法中：
代码语言：javascript复制protected Result doInvoke(final Invocation invocation) throws Throwable {
RpcInvocation inv = (RpcInvocation) invocation;
final String methodName = RpcUtils.getMethodName(invocation);
inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
inv.setAttachment(Constants.VERSION_KEY, version);
// 获取客户端连接对象
ExchangeClient currentClient;
if (clients.length == 1) {
currentClient = clients[0];
} else {
currentClient = clients[index.getAndIncrement() % clients.length];
}
try {
// 是否异步处理
boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
// 是否为单向通信，即只是客户端向服务端发送消息，不需要响应
boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY,Constants.DEFAULT_TIMEOUT);
if (isOneway) {
// 单向通信，因为不需要等待返回结果，所以这里应该是是异步的
boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
// 发送消息
currentClient.send(inv, isSent);
RpcContext.getContext().setFuture(null);
// 返回空结果
return new RpcResult();
} else if (isAsync) {
// 异步处理，并将返回结果设置到RPC上下文中，由用户调用ResponseFuture.get获取结果
ResponseFuture future = currentClient.request(inv, timeout) ;
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
// 同步等待结果返回
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
}
}该方法就是dubbo协议执行远程调用的逻辑，其中包含了异步和同步的处理逻辑，通过ResponseFuture实现，若异步则会将该类的实例设置到上下文环境中，由用户决定何时调用get方法获取结果；若同步则由框架自身调用get方法阻塞等待返回结果。ResponseFuture是一个接口，这里的默认实现是DefaultFuture，先来看看它是如何实现异步处理的：
代码语言：javascript复制// 响应结果
private volatile Response response;
// 接收结果，该方法由HeaderExchangeHandler调用
public static void received(Channel channel, Response response) {
try {
DefaultFuture future = FUTURES.remove(response.getId());
if (future != null) {
future.doReceived(response);
} else {
logger.warn("The timeout response finally returned at "
(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS").format(new Date()))
", response " response
(channel == null ? "" : ", channel: " channel.getLocalAddress()
" -> " channel.getRemoteAddress()));
}
} finally {
CHANNELS.remove(response.getId());
}
}
// 处理响应结果
private void doReceived(Response res) {
lock.lock();
try {
// 保存响应结果并唤醒阻塞线程（即get方法获取结果）
response = res;
if (done != null) {
done.signal();
}
} finally {
lock.unlock();
}
if (callback != null) {
invokeCallback(callback);
}
}
// 客户端获取响应结果
public Object get(int timeout) throws RemotingException {
if (timeout <= 0) {
timeout = Constants.DEFAULT_TIMEOUT;
}
// 还未接收到服务端的响应结果
if (! isDone()) {
long start = System.currentTimeMillis();
lock.lock();
try {
// 服务端还未响应完成时阻塞线程直到结果返回
while (! isDone()) {
done.await(timeout, TimeUnit.MILLISECONDS);
if (isDone() || System.currentTimeMillis() - start > timeout) {
break;
}
}
} catch (InterruptedException e) {
throw new RuntimeException(e);
} finally {
lock.unlock();
}
if (! isDone()) {
throw new TimeoutException(sent > 0, channel, getTimeoutMessage(false));
}
}
// 返回结果给用户
return returnFromResponse();
}
private Object returnFromResponse() throws RemotingException {
Response res = response;
if (res == null) {
throw new IllegalStateException("response cannot be null");
}
if (res.getStatus() == Response.OK) {
return res.getResult();
}
if (res.getStatus() == Response.CLIENT_TIMEOUT || res.getStatus() == Response.SERVER_TIMEOUT) {
throw new TimeoutException(res.getStatus() == Response.SERVER_TIMEOUT, channel, res.getErrorMessage());
}
throw new RemotingException(channel, res.getErrorMessage());
}上面几个方法为处理服务端响应的主要逻辑，用户可以通过该类的get方法获取响应结果，若是还未接受到响应结果就会阻塞线程直到获取到response，所以是异步还是同步就取决于get方法的调用时机。 我们继续看客户端的请求currentClient.request，但是这个client应该是什么对象呢？
这三个类应该不陌生的，我们在上文分析过，Dubbo默认使用共享连接，因此会使用ReferenceCountExchangeClient对其它两个进行装饰，提供引用计数的功能，这里不过多分析，向下最终都会进入到HeaderExchangeClient的request方法（LazyConnectExchangeClient只是实现懒加载功能，最终也是创建HeaderExchangeClient客户端）并继续调用HeaderExchangeChannel的request方法：
代码语言：javascript复制private final ExchangeChannel channel
public ResponseFuture request(Object request, int timeout) throws RemotingException {
return channel.request(request, timeout);
}
public ResponseFuture request(Object request, int timeout) throws RemotingException {
if (closed) {
throw new RemotingException(this.getLocalAddress(), null, "Failed to send request " request ", cause: The channel " this " is closed!");
}
// 封装请求参数对象
Request req = new Request();
req.setVersion("2.0.0");
// 双向通信标识
req.setTwoWay(true);
// 请求数据
req.setData(request);
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try{
// 使用netty发送消息
channel.send(req);
}catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}这里逻辑很清晰，也验证了上文的ResponseFuture对象实际为DefaultFuture，往下就是通过channel发送请求消息，由于Duboo默认使用的是Netty通信，所以这里应该是NettyChannel对象， 不过由于参数列表不匹配，这里首先是调用的其父类AbstractPeer的send方法，然后才进入到NettyChannel中：
代码语言：javascript复制// AbstractPeer
public void send(Object message) throws RemotingException {
send(message, url.getParameter(Constants.SENT_KEY, false));
}
// NettyChannel
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
ChannelFuture future = channel.write(message);
if (sent) {
timeout = getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
success = future.await(timeout);
}
Throwable cause = future.getCause();
if (cause != null) {
throw cause;
}
} catch (Throwable e) {
throw new RemotingException(this, "Failed to send message " message " to " getRemoteAddress() ", cause: " e.getMessage(), e);
}
if(! success) {
throw new RemotingException(this, "Failed to send message " message " to " getRemoteAddress()
"in timeout(" timeout "ms) limit");
}
}至此，客户端调用服务的流程就完成了，下面是整个客户端调用的时序图，为简略，省去了一些具体实现类：
由于是Netty通信，数据出站和入站时会有对应的编码和解码操作，这里就不分析了，下面直接看服务端如何接收处理请求的。
服务端接收请求
服务端是通过NettyHandler.messageReceived接收处理请求的：
代码语言：javascript复制private final ChannelHandler handler;
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
handler.received(channel, e.getMessage());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}主要逻辑在hander.received方法中，但这个handler有很多实现类，应该是哪一个呢？ 在分析发布服务的时候在HeaderExchanger.bind方法中包装了很多个handler：
代码语言：javascript复制public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}继续追踪Transporters.bind方法，最终会再对handler进行包装：
代码语言：javascript复制protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) {
return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class)
.getAdaptiveExtension().dispatch(handler, url)));
}这里拿到的扩展点为AllDispatcher，该类同样是包装handler：
代码语言：javascript复制public ChannelHandler dispatch(ChannelHandler handler, URL url) {
return new AllChannelHandler(handler, url);
}所以，最终的调用处理链为：
代码语言：javascript复制MultiMessageHandler
-> HeartbeatHandler
-> AllChannelHandler
-> DecodeHandler
-> HeaderExchangeHandler先来看看各个handler的作用：
- MultiMessageHandler ：复合消息处理
- HeartbeatHandler ：心跳消息的处理
- AllChannelHandler：线程派发处理器，将接收到的消息封装成ChannelEventRunnable交给线程池处理
- DecodeHandler：解码处理
- HeaderExchangeHandler：响应处理
前两个handler的逻辑没必要分析了，这里从AllChannelHandler开始，这个是Dubbo的线程派发模型的实现。什么是线程派发模型？简单的说就是Dubbo为提高请求处理效率，将能够简单快速处理的逻辑交由IO线程处理，而复杂耗时的操作交由线程池异步处理（哪些操作应放入线程池，可由用户配置决定）。Dubbo共提供了5种线程派发策略：
- all：所有消息都派发到线程池，包括请求，响应，连接事件，断开事件等
- direct：所有消息都不派发到线程池，全部在 IO 线程上直接执行
- message：只有请求和响应消息派发到线程池，其它消息均在 IO 线程上执行
- execution：只有请求消息派发到线程池，不含响应。其它消息均在 IO 线程上执行
- connection：在 IO 线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到线程池
默认情况下使用的是all，即AllChannelHandler：
代码语言：javascript复制public class AllChannelHandler extends WrappedChannelHandler {
// 连接消息
public void connected(Channel channel) throws RemotingException {
ExecutorService cexecutor = getExecutorService();
try{
cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.CONNECTED));
}catch (Throwable t) {
throw new ExecutionException("connect event", channel, getClass() " error when process connected event ." , t);
}
}
// 连接断开消息
public void disconnected(Channel channel) throws RemotingException {
ExecutorService cexecutor = getExecutorService();
try{
cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.DISCONNECTED));
}catch (Throwable t) {
throw new ExecutionException("disconnect event", channel, getClass() " error when process disconnected event ." , t);
}
}
// 接收响应消息
public void received(Channel channel, Object message) throws RemotingException {
ExecutorService cexecutor = getExecutorService();
try {
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
} catch (Throwable t) {
throw new ExecutionException(message, channel, getClass() " error when process received event .", t);
}
}
// 异常消息
public void caught(Channel channel, Throwable exception) throws RemotingException {
ExecutorService cexecutor = getExecutorService();
try{
cexecutor.execute(new ChannelEventRunnable(channel, handler ,ChannelState.CAUGHT, exception));
}catch (Throwable t) {
throw new ExecutionException("caught event", channel, getClass() " error when process caught event ." , t);
}
}
}当前流程我们主要看received方法，该方法将请求对象封装到ChannelEventRunnablec类中，而该类是实现了Runnable接口的，意为着我们直接看run方法就可以了：
代码语言：javascript复制public void run() {
switch (state) {
case RECEIVED:
try{
handler.received(channel, message);
}catch (Exception e) {
logger.warn("ChannelEventRunnable handle " state " operation error, channel is " channel
", message is " message,e);
}
break;
}
}这里我只截取了关键的代码片段，可以看到是直接交给下一个handler处理的，即DecodeHandler处理，该类是执行解码操作，也不分析了，直接看HeaderExchangeHandler.received方法：
代码语言：javascript复制public void received(Channel channel, Object message) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
if (message instanceof Request) {
// 处理请求
Request request = (Request) message;
if (request.isEvent()) {
// 处理事件消息
handlerEvent(channel, request);
} else {
// 处理普通请求
if (request.isTwoWay()) {
// 双向通信
Response response = handleRequest(exchangeChannel, request);
channel.send(response);
} else {
// 单向通信，无需返回请求结果
handler.received(exchangeChannel, request.getData());
}
}
} else if (message instanceof Response) {
// 处理响应
handleResponse(channel, (Response) message);
} else if (message instanceof String) {
.....
} else {
handler.received(exchangeChannel, message);
}
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException {
Response res = new Response(req.getId(), req.getVersion());
// 请求错误，返回错误消息
if (req.isBroken()) {
Object data = req.getData();
String msg;
if (data == null) msg = null;
else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data);
else msg = data.toString();
res.setErrorMessage("Fail to decode request due to: " msg);
res.setStatus(Response.BAD_REQUEST);
return res;
}
// 调用服务
Object msg = req.getData();
try {
// 继续调用，这里的handler又是哪一个呢？
Object result = handler.reply(channel, msg);
res.setStatus(Response.OK);
res.setResult(result);
} catch (Throwable e) {
res.setStatus(Response.SERVICE_ERROR);
res.setErrorMessage(StringUtils.toString(e));
}
return res;
}上面交互的逻辑不复杂，执行远程调用时会继续向下调用handler.reply方法，这个handler是什么呢？还记得在DubboProtocol.createServer方法中传入的requestHandler么？
代码语言：javascript复制private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
if (message instanceof Invocation) {
Invocation inv = (Invocation) message;
// 获取服务端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;
}
}
}
}
RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
// 调用invoke方法，最终会调用服务端实现的具体服务方法
return invoker.invoke(inv);
}
throw new RemotingException(channel, "Unsupported request: " message == null ? null : (message.getClass().getName() ": " message) ", channel: consumer: " channel.getRemoteAddress() " --> provider: " channel.getLocalAddress());
}
// 忽略其它方法
};通过getInvoker方法获取服务端的Invoker对象，然后通过该对象去调用具体的服务，但这里的Invoker对象具体是哪一个呢？在服务发布章节中，分析过Invoker，是通过JavassistProxyFactory.getInvoker创建了一个AbstractProxyInvoker的匿名内部类，并通过Wrapper去调用代理对象的方法，那这里获取到的就是这个匿名内部类么？ 还没有这么简单，在RegistryProtocol.doLocalExport方法中首先使用InvokerDelegete对Invoker进行了包装，然后调用protocol.export方法，这个之前我们也分析过，会经过ProtocolListenerWrapper -> ProtocolFilterWrapper -> DubboProtocol，主要看看filter.export：
代码语言：javascript复制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));
}
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.size() > 0) {
for (int i = filters.size() - 1; i >= 0; i --) {
final Filter filter = filters.get(i);
final Invoker<T> next = last;
last = new Invoker<T>() {
public Class<T> getInterface() {
return invoker.getInterface();
}
public URL getUrl() {
return invoker.getUrl();
}
public boolean isAvailable() {
return invoker.isAvailable();
}
public Result invoke(Invocation invocation) throws RpcException {
return filter.invoke(next, invocation);
}
public void destroy() {
invoker.destroy();
}
@Override
public String toString() {
return invoker.toString();
}
};
}
}
return last;
}可以看到这里获取了Filter扩展点构建了一个过滤链层层过滤，在配置文件中可以看到有如下过滤器：
代码语言：javascript复制echo=com.alibaba.dubbo.rpc.filter.EchoFilter
generic=com.alibaba.dubbo.rpc.filter.GenericFilter
genericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFilter
token=com.alibaba.dubbo.rpc.filter.TokenFilter
accesslog=com.alibaba.dubbo.rpc.filter.AccessLogFilter
activelimit=com.alibaba.dubbo.rpc.filter.ActiveLimitFilter
classloader=com.alibaba.dubbo.rpc.filter.ClassLoaderFilter
context=com.alibaba.dubbo.rpc.filter.ContextFilter
consumercontext=com.alibaba.dubbo.rpc.filter.ConsumerContextFilter
exception=com.alibaba.dubbo.rpc.filter.ExceptionFilter
executelimit=com.alibaba.dubbo.rpc.filter.ExecuteLimitFilter
deprecated=com.alibaba.dubbo.rpc.filter.DeprecatedFilter
compatible=com.alibaba.dubbo.rpc.filter.CompatibleFilter
timeout=com.alibaba.dubbo.rpc.filter.TimeoutFilter
monitor=com.alibaba.dubbo.monitor.support.MonitorFilter
validation=com.alibaba.dubbo.validation.filter.ValidationFilter
cache=com.alibaba.dubbo.cache.filter.CacheFilter
trace=com.alibaba.dubbo.rpc.protocol.dubbo.filter.TraceFilter
future=com.alibaba.dubbo.rpc.protocol.dubbo.filter.FutureFilter所以在requestHandler中调用invoker.invoke方法会经过上面的过滤器层层过滤后才到达最后的服务调用。
总结
Dubbo的服务发布、订阅以及交互过程是非常复杂的，总共用了三篇文章来理清整个流程，但其中还有很多的细节没有深入讲解，感兴趣的读者可自行分析，下一篇将分析Dubbo的服务目录源码，也是Dubbo系列的最后一篇。
