前言
经过前面这么长的篇幅,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修饰的属性会被注入一个代理对象
代理对象是用jdk动态代理生成的,肯定会调到InvocationHandler里,去执行bean的对应方法
创建的ReferenceBeanInvocationHandler
这个bean是dubbo创建的invoker对象,先是走到invoker对象的包装代理实例AbstractProxyInvoker中,由javassist代理invoker,调用他的invokeMethod方法,并把参数列表类型,远程接口,以及对应方法 作为参数传进去
服务降级
由于invoker对象先是经过集群容错ClusterInvoker的包装,会先调到MockClusterInvoker的invoker方法：根据的你的配置决定是否降级
在这里之前会把参数列表类型,远程接口,以及对应方法包装成一个invocation对象,可以理解成是一次调用信息的对象
集群容错 & 负载均衡
MockClusterInvoker对象里包装了其他集群容错Cluster实例,接下来会调用集群容错Cluster实例的invoker方法.
默认的集群容错策略是重试,对应FailoverClusterInvoker实例
会先调到父类的AbstractClusterInvoker的invoker方法,获取初始的服务列表和负载均衡策略,作为参数传入子类
第一次尝试调用,利用负载均衡算法,选择出具体的invoker对象,调他的invoker
过滤器链
这里获取到的invoker对象仍然是被过滤器链包装着的，
通过一个链表进行传递,链表中持有下一个过滤器的引用,调完自己的invoker之后,传给下一个过滤器调
可以看到是层层包装
最终调到dubboInvoker
netty发请求
dubboInvoker就会涉及到真正的rpc通信了
先是调到父类的invoke方法,往invocation里塞了一个attachment列表,是否远程接口名
再进入dubboInvoker的doInvoke方法
先是从之前已经创建好的netty连接客户端拿一个连接出来,判断你是异步,还是单工
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
双工(主要看这个)
双工的话是发了请求,等响应结果了再返回对应的结果
这里也是用的Future.get实现的,因为netty nio通信是异步的,这里用Future阻塞住,不停的去获取结果(一个变量),服务端响应了结果,然后去改这个变量的值,有了值才会返回,jdk里的Feature类也是这种做法。相当于 由异步转同步的效果。
这里的调用链路比较长,是初始化netty客户端创建的,可以看上篇.上图
HeaderExchangeClient.request
HeaderExchangeChannel.request,会把Invocation对象包装成一个Request对象
NettyClient.send(req)
先走父类AbstractPeer.send
再走父类AbstractClient.send(message,sent)
这里会判断你连接是否还有效,不行的话给你重新连接一下,获取一个新的channel对象
如果连接是有效的,就直接拿netty的channel对象
,钩到子类NettyClient.getChannel()
NettyChannel.getOrAddChannel(c, getUrl(), this)这行代码,会根据netty的channel对象与url与nettyClient对象包装成dubbo的NettyChannel对象,并建立缓存：netty的channel对象 ——> dubbo的NettyChannel对象
最后返回的就是一个包装了netty的channel对象与url与nettyClient对象的NettyChannel对象,调他的send方法
这个方法调的就是netty的channel对象的writeAndFlush发请求了。
还有个sent判断默认是false,是用来判断是否需要等待发送的结果的,默认不等待
为true的话就会用netty返回的Feature的await阻塞住当前线程,等待发送结果
最后看一下发送的消息体是啥样的
netty handler链的调用
根据netty的api,发消息之前会先调到pipeline注册的处理链中实现outBoundHandler接口的写方法
之前我们在启动netty服务端的时候,往netty的pipeline中注册了一个nettyClinetHanler,那么就会调到这个类实现ChannelDuplexHandler接口的write方法
nettyClinetHanler.write()
发送的时候不发生异常的话就会调到nettyClient的send方法(注册pipeline的时候new nettyClinetHanler传进来的)
之后就会走这个链路
nettyClient.send()啥也没干
MultiMessageHandler.send啥也没干
HeartbeatHandler.send 这里面会有发心跳的逻辑,设置了写的时间
AllChannelHandler.send 啥也没干
DecodeHandler.send 啥也没干
HeaderExchangeHandler.send 感觉也是没干啥
最后进到DubboPtotocol里的那个匿名的RequestHandler对象里面,也是调的父类的方法,啥也没干
走到这里,服务消费者也已经把调服务的请求发出去了。
最终返回了一个Feature对象
调他的get方法阻塞住,等待对端响应结果。
2.服务提供方响应请求
服务提供方启动netty服务端创建的handler链
也是一个requestHandler,和服务消费方的代码是公用的
getExchangerHandler也是HeaderExchanger对象
HeaderExchanger这次调的是bind方法
最里层包成这样了 DecodeHandler( HeaderExchangeHandler (requestHanle) )
又传到了Transporters.bind里面,又给套一个ChannelHandlerDispatcher
现在是ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanle) ))
再获取NettyTransporter实例，调把前面的传到nettyServer里去
NettyServer(ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanler) )))
进NettyServer构造方法,看看里面做了什么
乍一看,又开始包了,和服务消费方包的代码一模一样
最终是这样
HeaderExchangeServer(NettyServer(MultiMessageHandler( HeartbeatHandler(AllChannelHandler(ChannelHandlerDispatcher(DecodeHandler( HeaderExchangeHandler (requestHanler) ))))))
可以看到和服务消费方nettyClient的链差不多， nettyClient换成了NettyServer而已
NettyServer的构造方法回去开启netty服务端
先调父类构造
调回子类的doOpen
把NettyServer对象包到了NettyServerHandler中,注册到netty的pipeline的handler链中
最后调 bootstrap.bind(getBindAddress())
接受请求
接受请求的话,肯定是先从netty的pipeline的handler链开始的
注册到netty的pipeline的是NettyServerHandler,这个类肯定实现了netty的hanlder之类的接口,看起来和服务消费者的nettyClientHandler一样的继承关系,可以响应channel的读写操作
从netty调过来的NettyServerHandler.channelRead
又是一个缓存,和nettyClientHandler那边如出一辙，调的方法也一样
NettyServer.receive
MultiMessageHandler.receive
针对多消息体,返回多次结果
HeartbeatHandler.received
设置这次心跳读的时间，也不是专门响应心跳，而是响应业务请求,走下一个Handler
AllChannelHandler.received
这里是服务隔离思想，用的是线程池隔离,从线程池拿一个线程出来响应请求
run方法
DecodeHandler.received
这里会对消息体进行解码,把消息体弄成一个Request对象
HeaderExchangeHandler.received
走handleRequest会调用接口的具体方法
handleRequest()会调到DubboProtocol的那个RequestHandler匿名对象里
把message(request对象)弄成Invocation,根据Invocation对象获取invoker对象
然后调他的invoker方法
同样的,服务提供方调本地服务,也是要经过过滤器链的
这个链更长了
最后调到代理对象
由代理对象去调本地接口实例
最后层层返回到HeaderExchangeHandler.received方法,由于是双工通信需要返回给对端结果,所以会把结果以rpc通信的方式返回出去
注意这里的对象就是Response对象了
里头有个result
最后由用netty的channel对象,回写数据,这里走的就是和服务消费方发请求一样的逻辑了。
3.服务消费方接受响应结果
当服务提供方回写 响应之后之后,服务消费方就会收到响应结果的请求,和服务提供方接受请求是一样的逻辑
不同的是,这次接受的是个Response对象,不用回写数据给服务提供方
这个时候需要去唤醒之前等待结果的Feature里的线程了
先根据通信的id从缓存中移除并推出来对应的DefaultFuture对象
去唤醒DefaultFuture对象
对DefaultFuture中的response对象进行赋值,并唤醒
所以DefaultFuture的get方法会获取到返回值,并最终返回出去
最终就会成功获取到本地rpc调用的返回值,并打印
这就是一次完整的tcp调用的全部源码的流程了,end......
图解 ：
