服務提供者初始化完成以後，對外暴露Exporter。服務消費者初始化完成以後，獲得的是Proxy代理，方法調用的時候就是調用代理。web
服務消費者通過初始化以後，獲得的是一個動態代理類，InvokerInvocationHandler，包含MockClusterInvoker，MockClusterInvoker包含一個RegistryDirectory和FailoverClusterInvoker。緩存
Java動態代理，每個動態代理類都必需要實現InvocationHandler這個接口，而且每個代理類的實例都關聯到了一個handler，當咱們經過代理對象調用一個方法的時候，這個方法就會被轉發爲由實現了InvocationHandler這個接口的類的invoke方法來進行調用。app
服務消費者發起調用請求
InvokerInvocationHandler實現了InvocationHandler接口，當咱們調用helloService.sayHello();的時候，實際上會調用invoke()方法：負載均衡
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
|
//proxy是代理的真實對象
//method調用真實對象的方法
//args調用真實對象的方法的參數
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
//方法名sayHello
String methodName = method.getName();
//參數類型
Class<?>[] parameterTypes = method.getParameterTypes();
if (method.getDeclaringClass() == Object.class) {
return method.invoke(invoker, args);
}
if ("toString".equals(methodName) && parameterTypes.length == 0) {
return invoker.toString();
}
if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
return invoker.hashCode();
}
if ("equals".equals(methodName) && parameterTypes.length == 1) {
return invoker.equals(args[0]);
}
//invoker是MockClusterInvoker
//首先new RpcInvocation
//而後invoker.invoke
//最後recreate
//返回結果
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
|
先看下new RpcInvocation，Invocation是會話域，它持有調用過程當中的變量，好比方法名，參數類型等。dom
接着是invoker.invoke()，這裏invoker是MockClusterInvoker，進入MockClusterInvoker.invoker()：異步
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
public Result invoke(Invocation invocation) throws RpcException {
Result result = null;
//獲取mock屬性的值，咱們沒有配置，默認false
String value = directory.getUrl().getMethodParameter(invocation.getMethodName(), Constants.MOCK_KEY, Boolean.FALSE.toString()).trim();
if (value.length() == 0 || value.equalsIgnoreCase("false")){
//這裏invoker是FailoverClusterInvoker
result = this.invoker.invoke(invocation);
} else if (value.startsWith("force")) {
//force:direct mock
result = doMockInvoke(invocation, null);
} else {
//fail-mock
try {
result = this.invoker.invoke(invocation);
}catch (RpcException e) {
if (e.isBiz()) {
throw e;
} else {
result = doMockInvoke(invocation, e);
}
}
}
return result;
}
|
result = this.invoker.invoke(invocation);這裏invoker是FailoverClusterInvoker，會首先進入AbstractClusterInvoker的invoke方法：async
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
public Result invoke(final Invocation invocation) throws RpcException {
//檢查是否被銷燬
checkWheatherDestoried();
LoadBalance loadbalance;
//根據invocation中的參數來獲取全部的invoker列表
List<Invoker<T>> invokers = list(invocation);
if (invokers != null && invokers.size() > 0) {
//咱們沒有配置負載均衡的參數，默認使用random
//這裏獲得的是RandomLoadBalance
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);
}
//若是是異步操做默認添加invocation id
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
//這裏是子類實現，FailoverClusterInvoker中，執行調用
return doInvoke(invocation, invokers, loadbalance);
}
|
FailoverClusterInvoker.doInvoke()：ide
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
|
public Result doInvoke(Invocation invocation, final List<Invoker<T>> invokers, LoadBalance loadbalance) throws RpcException {
List<Invoker<T>> copyinvokers = invokers;
//檢查invokers是否爲空
checkInvokers(copyinvokers, invocation);
//重試次數
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.
//已經調用過的invoker
List<Invoker<T>> invoked = new ArrayList<Invoker<T>>(copyinvokers.size()); // invoked invokers.
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 {
//開始調用，返回結果
Result result = invoker.invoke(invocation);
return result;
} catch (RpcException e) {。。。 } finally {
providers.add(invoker.getUrl().getAddress());
}
}
throw new RpcException(。。。);
}
|
Result result = invoker.invoke(invocation);調用並返回結果，會首先進入InvokerWrapper，而後進入ListenerInvokerWrapper的invoke方法，接着進入AbstractInvoker的invoke：oop
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
public Result invoke(Invocation inv) throws RpcException {
if(destroyed) {
throw new RpcException(。。。);
}
//轉成RpcInvocation
RpcInvocation invocation = (RpcInvocation) inv;
invocation.setInvoker(this);
if (attachment != null && attachment.size() > 0) {
invocation.addAttachmentsIfAbsent(attachment);
}
Map<String, String> context = RpcContext.getContext().getAttachments();
if (context != null) {
invocation.addAttachmentsIfAbsent(context);
}
if (getUrl().getMethodParameter(invocation.getMethodName(), Constants.ASYNC_KEY, false)){
invocation.setAttachment(Constants.ASYNC_KEY, Boolean.TRUE.toString());
}
//異步的話，須要添加id
RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
try {
//這裏是DubboInvoker
return doInvoke(invocation);
} catch (InvocationTargetException e) { }
}
|
DubboInvoker.doInvoke()：ui
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
|
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;
//在初始化的時候，引用服務的過程當中會保存一個鏈接到服務端的Client
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) {//異步的，發送完須要獲得Future
ResponseFuture future = currentClient.request(inv, timeout) ;
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {//同步調用，咱們這裏使用的這種方式
RpcContext.getContext().setFuture(null);
//HeaderExchangeClient
return (Result) currentClient.request(inv, timeout).get();
}
} catch (TimeoutException e) {。。。}
}
|
咱們這裏使用的是同步調用，看(Result) currentClient.request(inv, timeout).get();方法，這裏的client是ReferenceCountExchangeClient，直接調用HeaderExchangeClient的request方法：
1
2
3
4
|
public ResponseFuture request(Object request, int timeout) throws RemotingException {
//這裏的Channel是HeaderExchangeChannel
return channel.request(request, timeout);
}
|
進入HeaderExchangeChannel的request方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
public ResponseFuture request(Object request, int timeout) throws RemotingException {
if (closed) {
throw new RemotingException(。。。);
}
//建立一個請求頭
Request req = new Request();
req.setVersion("2.0.0");
req.setTwoWay(true);
//這裏request參數裏面保存着
//methodName = "sayHello"
//parameterTypes = {Class[0]@2814}
//arguments = {Object[0]@2768}
//attachments = {HashMap@2822} size = 4
//invoker = {DubboInvoker@2658}
req.setData(request);
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try{
//這裏的channel是NettyClient
//發送請求
channel.send(req);
}catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
|
channel.send(req)，首先會調用AbstractPeer的send方法：
1
2
3
4
|
//子類處理，接着是AbstractClient執行發送
public void send(Object message) throws RemotingException {
send(message, url.getParameter(Constants.SENT_KEY, false));
}
|
AbstractClient執行發送：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
public void send(Object message, boolean sent) throws RemotingException {
//重連
if (send_reconnect && !isConnected()){
connect();
}
//先獲取Channel，是在NettyClient中實現的
Channel channel = getChannel();
//TODO getChannel返回的狀態是否包含null須要改進
if (channel == null || ! channel.isConnected()) {
throw new RemotingException(this, "message can not send, because channel is closed . url:" + getUrl());
}
channel是NettyChannel
channel.send(message, sent);
}
|
channel.send(message, sent);首先通過AbstractChannel的send方法處理，只是判斷是否關閉了，而後是NettyChannel的send來繼續處理，這裏就把消息發送到服務端了：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
public void send(Object message, boolean sent) throws RemotingException {
super.send(message, sent);
boolean success = true;
int timeout = 0;
try {
//交給netty處理
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的處理後會到dubbo中的codec相關方法中先進行解碼，入口是NettyCodecAdapter.messageReceived()，關於這個方法的代碼在dubbo編解碼的那篇文章中已經分析過，再也不重複。通過解碼以後，會進入到NettyHandler.messageReceived()方法：
1
2
3
4
5
6
7
8
9
10
|
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
//獲取channel
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
//這裏handler是NettyServer
handler.received(channel, e.getMessage());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
|
接着會進入AbstractPeer的received方法：
1
2
3
4
5
6
7
|
public void received(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
//這裏是MultiMessageHandler
handler.received(ch, msg);
}
|
進入MultiMessageHandler的received方法：
1
2
3
4
5
6
7
8
9
10
11
12
|
public void received(Channel channel, Object message) throws RemotingException {
//是多消息的話，使用多消息處理器處理
if (message instanceof MultiMessage) {
MultiMessage list = (MultiMessage)message;
for(Object obj : list) {
handler.received(channel, obj);
}
} else {
//這裏是HeartbeatHandler
handler.received(channel, message);
}
}
|
進入HeartbeatHandler的received方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
public void received(Channel channel, Object message) throws RemotingException {
setReadTimestamp(channel);
//心跳請求處理
if (isHeartbeatRequest(message)) {
Request req = (Request) message;
if (req.isTwoWay()) {
Response res = new Response(req.getId(), req.getVersion());
res.setEvent(Response.HEARTBEAT_EVENT);
channel.send(res);
}
return;
}
//心跳回應消息處理
if (isHeartbeatResponse(message)) {
return;
}
//這裏是AllChannelHandler
handler.received(channel, message);
}
|
繼續進入AllChannelHandler的received方法：
1
2
3
4
5
6
7
8
|
public void received(Channel channel, Object message) throws RemotingException {
//獲取線程池執行
ExecutorService cexecutor = getExecutorService();
try {
//handler是DecodeHandler
cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
} catch (Throwable t) { }
}
|
這裏會去啓動新線程執行ChannelEventRunnable的run方法，接着去調用DecodeHandler的received方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
public void received(Channel channel, Object message) throws RemotingException {
//不清楚啥意思
if (message instanceof Decodeable) {
decode(message);
}
//解碼請求類型
if (message instanceof Request) {
decode(((Request)message).getData());
}
//解碼響應類型
if (message instanceof Response) {
decode( ((Response)message).getResult());
}
//解碼以後到HeaderExchangeHandler中處理
handler.received(channel, message);
}
|
解碼以後到HeaderExchangeHandler的received方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
public void received(Channel channel, Object message) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
//request類型的消息
if (message instanceof Request) {
Request request = (Request) message;
if (request.isEvent()) {//判斷心跳仍是正常請求
// 處理心跳
handlerEvent(channel, request);
} else {//正常的請求
//須要返回
if (request.isTwoWay()) {
//處理請求，並構造響應信息
Response response = handleRequest(exchangeChannel, request);
//NettyChannel，發送響應信息
channel.send(response);
} else {//不須要返回的處理
handler.received(exchangeChannel, request.getData());
}
}
} else if (message instanceof Response) {//response類型的消息
handleResponse(channel, (Response) message);
} else if (message instanceof String) {
if (isClientSide(channel)) {
Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
} else {//telnet類型
String echo = handler.telnet(channel, (String) message);
if (echo != null && echo.length() > 0) {
channel.send(echo);
}
}
} else {
handler.received(exchangeChannel, message);
}
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
|
先看下處理請求，並構造響應信息：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
|
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;
}
// find handler by message class.
Object msg = req.getData();
try {
//處理請求數據，handler是DubboProtocol中的new的一個ExchangeHandlerAdapter
Object result = handler.reply(channel, msg);
res.setStatus(Response.OK);
res.setResult(result);
} catch (Throwable e) { }
return res;
}
|
進入DubboProtocol中的ExchangeHandlerAdapter的replay方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
|
public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
if (message instanceof Invocation) {
//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;
}
}
}
if (!hasMethod){
return null;
}
}
RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
//執行調用，而後返回結果
return invoker.invoke(inv);
}
throw new RemotingException(。。。);
}
|
先看下getInvoker獲取Invoker：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
|
Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException{
boolean isCallBackServiceInvoke = false;
boolean isStubServiceInvoke = false;
int port = channel.getLocalAddress().getPort();
String path = inv.getAttachments().get(Constants.PATH_KEY);
//若是是客戶端的回調服務.
isStubServiceInvoke = Boolean.TRUE.toString().equals(inv.getAttachments().get(Constants.STUB_EVENT_KEY));
if (isStubServiceInvoke){
port = channel.getRemoteAddress().getPort();
}
//callback
isCallBackServiceInvoke = isClientSide(channel) && !isStubServiceInvoke;
if(isCallBackServiceInvoke){
path = inv.getAttachments().get(Constants.PATH_KEY)+"."+inv.getAttachments().get(Constants.CALLBACK_SERVICE_KEY);
inv.getAttachments().put(IS_CALLBACK_SERVICE_INVOKE, Boolean.TRUE.toString());
}
String serviceKey = serviceKey(port, path, inv.getAttachments().get(Constants.VERSION_KEY), inv.getAttachments().get(Constants.GROUP_KEY));
//從以前緩存的exporterMap中查找Exporter
//key：dubbo.common.hello.service.HelloService:20880
DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
if (exporter == null)
throw new RemotingException(。。。）;
//獲得Invoker，返回
return exporter.getInvoker();
}
|
再看執行調用invoker.invoke(inv);，會先進入InvokerWrapper：
1
2
3
|
public Result invoke(Invocation invocation) throws RpcException {
return invoker.invoke(invocation);
}
|
接着進入AbstractProxyInvoker：
1
2
3
4
5
6
7
|
public Result invoke(Invocation invocation) throws RpcException {
try {
//先doInvoke
//而後封裝成結果返回
return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
} catch (InvocationTargetException e) {。。。}
}
|
這裏的doInvoke是在JavassistProxyFactory中的AbstractProxyInvoker實例：
1
2
3
4
5
6
7
8
9
10
11
12
13
|
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
// TODO Wrapper類不能正確處理帶$的類名
final Wrapper wrapper = Wrapper.getWrapper(proxy.getClass().getName().indexOf('$') < 0 ? proxy.getClass() : type);
return new AbstractProxyInvoker<T>(proxy, type, url) {
@Override
protected Object doInvoke(T proxy, String methodName,
Class<?>[] parameterTypes,
Object[] arguments) throws Throwable {
//這裏就調用了具體的方法
return wrapper.invokeMethod(proxy, methodName, parameterTypes, arguments);
}
};
}
|
消息處理完後返回到HeaderExchangeHandler的received方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
|
public void received(Channel channel, Object message) throws RemotingException {
channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
try {
//request類型的消息
if (message instanceof Request) {
Request request = (Request) message;
if (request.isEvent()) {//判斷心跳仍是正常請求
// 處理心跳
handlerEvent(channel, request);
} else {//正常的請求
//須要返回
if (request.isTwoWay()) {
//處理請求，並構造響應信息，這在上面已經解析過了
Response response = handleRequest(exchangeChannel, request);
//NettyChannel，發送響應信息
channel.send(response);
} else {//不須要返回的處理
handler.received(exchangeChannel, request.getData());
}
}
} else if (message instanceof Response) {//response類型的消息
handleResponse(channel, (Response) message);
} else if (message instanceof String) {
if (isClientSide(channel)) {
Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
} else {//telnet類型
String echo = handler.telnet(channel, (String) message);
if (echo != null && echo.length() > 0) {
channel.send(echo);
}
}
} else {
handler.received(exchangeChannel, message);
}
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
|
解析完請求，構造完響應消息，就開始發送響應了,channel.send(response);，先通過AbstractPeer:
1
2
3
4
|
public void send(Object message) throws RemotingException {
//NettyChannel
send(message, url.getParameter(Constants.SENT_KEY, false));
}
|
進入NettyChannel中，進行響應消息的發送：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
|
public void send(Object message, boolean sent) throws RemotingException {
//AbstractChannel的處理
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");
}
}
|
消費者接受到服務端返回的響應後的處理
服務提供者端接收到消費者端的請求並處理以後，返回給消費者端，消費者這邊接受響應的入口跟提供者差很少，也是NettyCodecAdapter.messageReceived()，通過解碼，到NettyHandler.messageReceived()處理：
1
2
3
4
5
6
7
8
9
|
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
try {
//NettyClient
handler.received(channel, e.getMessage());
} finally {
NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
}
}
|
先通過AbstractPeer的received方法：
1
2
3
4
5
6
7
|
public void received(Channel ch, Object msg) throws RemotingException {
if (closed) {
return;
}
//MultiMessageHandler
handler.received(ch, msg);
}
|
進入MultiMessageHandler：
1
2
3
4
5
6
7
8
9
10
11
|
public void received(Channel channel, Object message) throws RemotingException {
if (message instanceof MultiMessage) {
MultiMessage list = (MultiMessage)message;
for(Object obj : list) {
handler.received(channel, obj);
}
} else {
//HeartbeatHandler
handler.received(channel, message);
}
}
|
進入HeartbeatHandler，根據不一樣類型進行處理：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
|
public void received(Channel channel, Object message) throws RemotingException {
setReadTimestamp(channel);
if (isHeartbeatRequest(message)) {
Request req = (Request) message;
if (req.isTwoWay()) {
Response res = new Response(req.getId(), req.getVersion());
res.setEvent(Response.HEARTBEAT_EVENT);
channel.send(res);
if (logger.isInfoEnabled()) {
int heartbeat = channel.getUrl().getParameter(Constants.HEARTBEAT_KEY, 0);
if(logger.isDebugEnabled()) {
logger.debug("Received heartbeat from remote channel " + channel.getRemoteAddress()
+ ", cause: The channel has no data-transmission exceeds a heartbeat period"
+ (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
}
}
}
return;
}
if (isHeartbeatResponse(message)) {
if (logger.isDebugEnabled()) {
logger.debug(
new StringBuilder(32)
.append("Receive heartbeat response in thread ")
.append(Thread.currentThread().getName())
.toString());
}
return;
}
//AllChannelHandler
handler.received(channel, message);
}
|
進入AllChannelHandler：
1
2
3
4
5
6
7
8
|
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);
}
}
|
而後在新線程，ChannelEventRunnable的run方法中進入DecodeHandler：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
|
public void received(Channel channel, Object message) throws RemotingException {
if (message instanceof Decodeable) {
decode(message);
}
if (message instanceof Request) {
decode(((Request)message).getData());
}
//這裏進行response類型的處理
if (message instanceof Response) {
decode( ((Response)message).getResult());
}
handler.received(channel, message);
}
|
進入處理response的decode方法，進行解碼response：
1
2
3
4
5
6
7
|
private void decode(Object message) {
if (message != null && message instanceof Decodeable) {
try {
((Decodeable)message).decode();
} catch (Throwable e) {。。。} // ~ end of catch
} // ~ end of if
}
|
接着會進入HeaderExchangerHandler.received () 方法：
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
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) {
//這裏處理response消息
handleResponse(channel, (Response) message);
} else if (message instanceof String) {
if (isClientSide(channel)) { Exception } else {
String echo = handler.telnet(channel, (String) message);
if (echo != null && echo.length() > 0) {
channel.send(echo);
}
}
} else {
handler.received(exchangeChannel, message);
}
} finally {
HeaderExchangeChannel.removeChannelIfDisconnected(channel);
}
}
|
handleResponse方法：
1
2
3
4
5
|
static void handleResponse(Channel channel, Response response) throws RemotingException {
if (response != null && !response.isHeartbeat()) {
DefaultFuture.received(channel, response);
}
}
|
這一步設置response到消費者請求的Future中，以供消費者經過DefaultFuture.get()取得提供者的響應，此爲同步轉異步重要一步，且請求超時也由DefaultFuture控制。
而後就是return (Result) currentClient.request(inv, timeout).get();在DubboInvoker中，這裏繼續執行，而後執行Filter，最後返回到InvokerInvocationHandler.invoker()方法中，方法獲得調用結果，結束！
注意：
消費者端的DubboInvoker發起請求後，後續的邏輯是異步的或是指定超時時間內阻塞的，直到獲得響應結果後，繼續執行DubboInvoker中邏輯。
對於異步請求時，消費者獲得Future，其他邏輯均是異步的。
消費者還能夠經過設置async、sent、return來調整處理邏輯，async指異步仍是同步請求，sent指是否等待請求消息發出即阻塞等待是否成功發出請求、return指是否忽略返回值即但方向通訊，通常異步時使用以減小Future對象的建立和管理成本。
