1、consumer 发送扩展
我们先来看一下 dubbo 中 consumer
端的请求发送原理,也就是从 InvokerWrapper#invoke 开始,在 consumer 服务引用分析的时候,我们知道根据 Invoke 调用的时候, dubbo 会创建 ProtocolListenerWrapper
与 ProtocolFilterWrapper 来用集成框架使用者的扩展包含:InvokerListener
与 Filter
。ProtocolListenerWrapper
在对象创建的时候就会调用InvokerListener#referred
扩展,所以在远程服务调用的时候最主要的还是 Filter
扩展(ConsumerContextFilter FutureFilter MonitorFilter)
2、consumer 发送原理
最终 consumer
会到 DubboInvoke
进行服务调用。它会在 AbstractInvoker#invoke
添加一些扩展参数到 RpcInvocation
这个远程调用对象里面。添加的扩展参数包含:
- interface : 远程调用的接口名称
- group : 接口分组名称
- token : 调用的 token 信息
- timeout : 调用服务的超时时间
- async : 是否异步调用
- id : 异步操作默认添加 invocation id,用于保证操作幂等
以及 RpcContext
传递过来的扩展参数(RpcContext#attachments
)。然后在 DubboInvoker#doInvoke
中会添加 path
(接口全类名) 以及 version
(版本信息)。再根据 dubbo 的调用模式进行远程调用,包含以下三种调用模式:
- oneway 模式:
<dubbo:method>
标签的return
属性配置为false,则是oneway模式,利用ExchangeClient 对象向服务端发送请求消息之后,立即返回空 RpcResult 对象 - 异步模式:
<dubbo:method>
标签的async
属性配置为 ture,则是异步模式,直接返回空RpcResult
对象,由FutureFilter
和DefaultFuture
完成异步处理工作 - 同步模式:默认即是同步,则发送请求之后线程进入等待状态,直到收到服务端的响应消息或者超时。
下面我们看一下 dubbo 同步调用时序图:
ChannelFuture future = channel.write(message);
最终是调用 org.jboss.netty.channel.Channel
通过 socket 发送消息到从集群中选择出的一个暴露服务信息的服务器发送网络数据。
3、consumer 接收原理
我们都知道 dubbo 其实是通过 netty
来进行 socket 通信的。而在使用 netty
进行网络编程的时候,其实核心就是就是实现 ChannelHandler
。而在 dubbo 中对应的实现类就是 NettyHandler
(高版本支持支持 netty 4
使用的是 NettyClientHandler
,NettyHandler 使用的是 netty 3.x
)。如果在 consumer
端(provider
也支持)需要使用 netty 4
进行业务处理,需要进行进行以下配置:
<dubbo:consumer client="netty4" />
所以 consumer
接收 provider
响应的入口就在 NettyClientHandler#channelRead
:
首先 ChannelHandler
用于接收 provider
端响应回来的请求,然后经过 3 个 dubbo 自定义的 ChannelHandler
。
- MultiMessageHandler:支持
MultiMessage
消息处理,也就是多条消息处理。 - HeartbeatHandler:
netty
心条检测。如果心跳请求,发送心跳然后直接return
,如果是心跳响应直接return
。 - DecodeHandler:解码
message
,解析成 dubbo 中的Response
对象 - HeaderExchangeHandler:处理解析后的
provider
端返回的Response
响应信息,把响应结果赋值到DefaultFuture
响应获取阻塞对象中。
4、dubbo 异步变同步
我们都知道 dubbo 是基于 netty NIO
的非阻塞 并行调用通信。所以 dubbo 在 consumer
请求 provider
后响应都是异步的。但是在 dubbo 里面默认是同步返回的,那么 dubbo 是如何把异步响应变成同步请求的呢?带着这个问题,首先我们来看一下 dubbo 里面的几种请求方式。
4.1 异步且无返回值
这种请求最简单,consumer
把请求信息发送给 provider
就行了。只是需要在 consumer
端把请求方式配置成异步请求就好了。如下:
<dubbo:method name="sayHello" return="false"></dubbo:method>
4.2 异步且有返回值
consumer
首先把请求信息发送给 provider
。这个时候在 consumer
端不仅把请求方式配置成异步,并且需要 RpcContext
这个 ThreadLocal
对象获取到 Future
对象,然后通过 Future#get()
阻塞式获取到 provider
的响应。
那么这个 Future
是如果添加到 RpcContext
中呢?
ResponseFuture future = currentClient.request(inv, timeout);
FutureAdapter<T> futureAdapter = new FutureAdapter<>(future);
RpcContext.getContext().setFuture(futureAdapter);
Result result;
if (RpcUtils.isAsyncFuture(getUrl(), inv)) {
result = new AsyncRpcResult<>(futureAdapter);
} else {
result = new RpcResult();
}
return result;
4.3 异步变同步(默认)
下面我们就来讨论一下 dubbo 是如何把异步请求转化成同步请求的。其实原理和异步请求的通过 Future#get
等待 provider
响应返回一样,只不过异步有返回值是显示调用而默认是 dubbo 内部把这步完成了。(当前线程怎么让它 “暂停”,等结果回来后,再执行?)
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!");
}
// create request.
Request req = new Request();
req.setVersion(Version.getProtocolVersion());
req.setTwoWay(true);
req.setData(request);
//通过 dubbo 自定义的 Channel、Request 与 timeout(int) 构造一个 DefaultFuture 对象。
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try {
channel.send(req);
} catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
public class DefaultFuture implements ResponseFuture {
private static final Map<Long, Channel> CHANNELS = new ConcurrentHashMap<Long, Channel>();
private static final Map<Long, DefaultFuture> FUTURES = new ConcurrentHashMap<Long, DefaultFuture>();
private final long id;
private final Channel channel;
private final Request request;
private final int timeout;
public DefaultFuture(Channel channel, Request request, int timeout) {
this.channel = channel;
this.request = request;
this.id = request.getId();//id 是在创建 Request 的时候使用 AtomicLong#getAndIncrement 生成的。从 1 开始并且如果它一直增加直到生成负数也能保证这台机器这个值是唯一的,且不冲突的。符合唯一主键原则
this.timeout = timeout > 0 ? timeout : channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
// put into waiting map.
FUTURES.put(id, this);
CHANNELS.put(id, channel);
}
}
ResponseFuture#get 方法上面,下面我们来看一下这个方法的源码:
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();
}
其实就是 while 循环,利用 java 的 lock
机制判断如果在超时时间范围内 DefaultFuture#response
如果赋值成不为空就返回响应,否则抛出 TimeoutException
异常。下面我们就来看一下 DefaultFuture#response
是如何被赋值的。
还记得 consumer
接收 provider
响应的最后一步吗?就是 DefaultFuture#received
,在 provider
端会带回 consumer
请求的 id
。我们来看一下它的具体处理逻辑:
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());
}
}
它会从最开始通过构造函数传进去的 DefaultFuture#FUTURES
根据请求的 id
拿到 DefaultFuture
,然后根据这个 DefaultFuture
调用 DefaultFuture#doReceived
方法。通过 Java 里面的 lock
机制把 provider
的值赋值给 DefaultFuture#response
。此时 consumer
也正在调用 DefaultFuture#get
方法进行阻塞,当这个 DefaultFuture#response
被赋值后,它的值就不为空。阻塞操作完成,且根据请求号的 id
把 consumer
端的 Request
以及 Provider
端返回的 Response
关联了起来。