【dubbo源码】22. 一次完整的rpc调用

前言

经过前面这么长的篇幅,rpc通信的所有的准备工作dubbo都已经在启动的时候准备就绪。

服务暴露方:

• 将自己注册到注册中心
• 开启了netty服务端就用接受请求

服务消费方:

• 将自己注册到注册中心
• 获取到了某个服务的服务列表
• 持有了某个服务的某台主机的invoker对象,并针对其url建立了netty服务端,并发起了连接。

rpc通信流程

• 服务消费方 向服务暴露方 发起tcp请求
  • 服务降级
  • 集群容错
  • 负载均衡
  • 过滤器
  • netty发请求
• 服务提供方接到请求后,调用指定的服务方法,并响应数据
  • nettyHandler响应channelRead
  • 调用本地方法
  • 回写数据
• 服务消费方接到服务提供的响应数据后进行业务处理
  • 接受响应结果(Response对象不回写数据)
  • 唤醒阻塞线程
  • 最终得到结果

测试代码：rpc调用另外一个服务的getUsername方法,并传一个string参数

    public static void main(String[] args) throws IOException {
    AnnotationConfigApplicationContext applicationContext = new AnnotationConfigApplicationContext(AnnoBean.class);
    OrderServiceImpl orderService = (OrderServiceImpl) applicationContext.getBean("orderServiceImpl");
    orderService.test();
    System.in.read();
}

OrderService

@Service

public class OrderServiceImpl implements OrderService {

@Reference(mock = "true")
private UserService userService;

@Override
public void test() {
    System.out.println(userService.getUsername("liuben"));
}

}

服务暴露方UserService

@Service
public class UserServiceImpl implements     UserService {
    @Override
    public String getUsername(String username)     {
        return "i am "+username;
    }
}

1. 服务消费方发起tcp请求

@Reference修饰的属性会被注入一个代理对象

image

代理对象是用jdk动态代理生成的,肯定会调到InvocationHandler里,去执行bean的对应方法

image

创建的ReferenceBeanInvocationHandler

image

这个bean是dubbo创建的invoker对象,先是走到invoker对象的包装代理实例AbstractProxyInvoker中,由javassist代理invoker,调用他的invokeMethod方法,并把参数列表类型,远程接口,以及对应方法 作为参数传进去

image

服务降级

由于invoker对象先是经过集群容错ClusterInvoker的包装,会先调到MockClusterInvoker的invoker方法：根据的你的配置决定是否降级

在这里之前会把参数列表类型,远程接口,以及对应方法包装成一个invocation对象,可以理解成是一次调用信息的对象

image

集群容错 & 负载均衡

MockClusterInvoker对象里包装了其他集群容错Cluster实例,接下来会调用集群容错Cluster实例的invoker方法.

默认的集群容错策略是重试,对应FailoverClusterInvoker实例

会先调到父类的AbstractClusterInvoker的invoker方法,获取初始的服务列表和负载均衡策略,作为参数传入子类

image

第一次尝试调用,利用负载均衡算法,选择出具体的invoker对象,调他的invoker

image

过滤器链

这里获取到的invoker对象仍然是被过滤器链包装着的，

image

通过一个链表进行传递,链表中持有下一个过滤器的引用,调完自己的invoker之后,传给下一个过滤器调

image

可以看到是层层包装

image

最终调到dubboInvoker

netty发请求

dubboInvoker就会涉及到真正的rpc通信了

先是调到父类的invoke方法,往invocation里塞了一个attachment列表,是否远程接口名

image

再进入dubboInvoker的doInvoke方法

先是从之前已经创建好的netty连接客户端拿一个连接出来,判断你是异步,还是单工

image

isOneway 单工
单工就是直接返回默认的结果,不需要获取返回值

if (isOneway) {
    boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
    currentClient.send(inv, isSent);
            RpcContext.getContext().setFuture(null);
    return new RpcResult();
    //@Reference(chefalse,methods = {@Method(= "asynctoDo",async = tru异步调用
} 

isAsync 异步

异步的话,其实就是先返回结果,用一个Feature阻塞住,不停的去拿结果,拿到结果再响应逻辑

else if (isAsync) {
    ResponseFuture futurecurrentClient.request(intimeout);
    Context.getContext().seture(nFutureAdapter<Object>(futur;
    return new RpcResult();
}

Feature对象.get

image

双工(主要看这个)

双工的话是发了请求,等响应结果了再返回对应的结果

这里也是用的Future.get实现的,因为netty nio通信是异步的,这里用Future阻塞住,不停的去获取结果(一个变量),服务端响应了结果,然后去改这个变量的值,有了值才会返回,jdk里的Feature类也是这种做法。相当于 由异步转同步的效果。

image

这里的调用链路比较长,是初始化netty客户端创建的,可以看上篇.上图

image

HeaderExchangeClient.request

image

HeaderExchangeChannel.request,会把Invocation对象包装成一个Request对象

image

NettyClient.send(req)

先走父类AbstractPeer.send

image

再走父类AbstractClient.send(message,sent)
这里会判断你连接是否还有效,不行的话给你重新连接一下,获取一个新的channel对象

image

如果连接是有效的,就直接拿netty的channel对象
,钩到子类NettyClient.getChannel()

image

NettyChannel.getOrAddChannel(c, getUrl(), this)这行代码,会根据netty的channel对象与url与nettyClient对象包装成dubbo的NettyChannel对象,并建立缓存：netty的channel对象 ——> dubbo的NettyChannel对象

image

最后返回的就是一个包装了netty的channel对象与url与nettyClient对象的NettyChannel对象,调他的send方法

image

这个方法调的就是netty的channel对象的writeAndFlush发请求了。

还有个sent判断默认是false,是用来判断是否需要等待发送的结果的,默认不等待

为true的话就会用netty返回的Feature的await阻塞住当前线程,等待发送结果

最后看一下发送的消息体是啥样的

image

netty handler链的调用

根据netty的api,发消息之前会先调到pipeline注册的处理链中实现outBoundHandler接口的写方法

之前我们在启动netty服务端的时候,往netty的pipeline中注册了一个nettyClinetHanler,那么就会调到这个类实现ChannelDuplexHandler接口的write方法

image

nettyClinetHanler.write()

image

发送的时候不发生异常的话就会调到nettyClient的send方法(注册pipeline的时候new nettyClinetHanler传进来的)

之后就会走这个链路

image

nettyClient.send()啥也没干

image

MultiMessageHandler.send啥也没干

image

HeartbeatHandler.send 这里面会有发心跳的逻辑,设置了写的时间

image

AllChannelHandler.send 啥也没干

image

DecodeHandler.send 啥也没干

image

HeaderExchangeHandler.send 感觉也是没干啥

image

最后进到DubboPtotocol里的那个匿名的RequestHandler对象里面,也是调的父类的方法,啥也没干

image

走到这里,服务消费者也已经把调服务的请求发出去了。

最终返回了一个Feature对象

image

调他的get方法阻塞住,等待对端响应结果。

image image

2.服务提供方响应请求

服务提供方启动netty服务端创建的handler链

image

也是一个requestHandler,和服务消费方的代码是公用的

image

getExchangerHandler也是HeaderExchanger对象

image

HeaderExchanger这次调的是bind方法

image

最里层包成这样了 DecodeHandler( HeaderExchangeHandler (requestHanle) )

又传到了Transporters.bind里面,又给套一个ChannelHandlerDispatcher

现在是ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanle) ))

image

再获取NettyTransporter实例，调把前面的传到nettyServer里去
NettyServer(ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanler) )))

image

进NettyServer构造方法,看看里面做了什么

乍一看,又开始包了,和服务消费方包的代码一模一样

image

最终是这样
HeaderExchangeServer(NettyServer(MultiMessageHandler( HeartbeatHandler(AllChannelHandler(ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanler) ))))))
可以看到和服务消费方nettyClient的链差不多， nettyClient换成了NettyServer而已

NettyServer的构造方法回去开启netty服务端

先调父类构造

image

调回子类的doOpen

把NettyServer对象包到了NettyServerHandler中,注册到netty的pipeline的handler链中

image

最后调 bootstrap.bind(getBindAddress())

接受请求

接受请求的话,肯定是先从netty的pipeline的handler链开始的

注册到netty的pipeline的是NettyServerHandler,这个类肯定实现了netty的hanlder之类的接口,看起来和服务消费者的nettyClientHandler一样的继承关系,可以响应channel的读写操作

image

从netty调过来的NettyServerHandler.channelRead

image

又是一个缓存,和nettyClientHandler那边如出一辙，调的方法也一样

image

NettyServer.receive

image

MultiMessageHandler.receive

针对多消息体,返回多次结果

image

HeartbeatHandler.received

设置这次心跳读的时间，也不是专门响应心跳，而是响应业务请求,走下一个Handler

image

AllChannelHandler.received

这里是服务隔离思想，用的是线程池隔离,从线程池拿一个线程出来响应请求

image

run方法

image

DecodeHandler.received

这里会对消息体进行解码,把消息体弄成一个Request对象

image image

HeaderExchangeHandler.received

走handleRequest会调用接口的具体方法

image

handleRequest()会调到DubboProtocol的那个RequestHandler匿名对象里

image image

把message(request对象)弄成Invocation,根据Invocation对象获取invoker对象

image image

然后调他的invoker方法

image

同样的,服务提供方调本地服务,也是要经过过滤器链的

image

这个链更长了

image

最后调到代理对象

image

由代理对象去调本地接口实例

image

最后层层返回到HeaderExchangeHandler.received方法,由于是双工通信需要返回给对端结果,所以会把结果以rpc通信的方式返回出去

image

注意这里的对象就是Response对象了

里头有个result

image

最后由用netty的channel对象,回写数据,这里走的就是和服务消费方发请求一样的逻辑了。

image

3.服务消费方接受响应结果

当服务提供方回写 响应之后之后,服务消费方就会收到响应结果的请求,和服务提供方接受请求是一样的逻辑
不同的是,这次接受的是个Response对象,不用回写数据给服务提供方

image

这个时候需要去唤醒之前等待结果的Feature里的线程了

image

先根据通信的id从缓存中移除并推出来对应的DefaultFuture对象

image

去唤醒DefaultFuture对象

image

对DefaultFuture中的response对象进行赋值,并唤醒

image

所以DefaultFuture的get方法会获取到返回值,并最终返回出去

image image

最终就会成功获取到本地rpc调用的返回值,并打印

image

这就是一次完整的tcp调用的全部源码的流程了,end......

图解 ：

image

dubbo rpc时序图