一般情况，consumer发送请求时，创建一个DefaultFuture对象，然后阻塞并等待响应。DefaultFuture类，封装了请求和响应：
// 省略其他代码
public class DefaultFuture implements ResponseFuture {
private final Lock lock = new ReentrantLock();
private final Condition done = lock.newCondition();
//waiting map
private static final Map<Long, DefaultFuture> FUTURES
= new ConcurrentHashMap<Long, DefaultFuture>();
private final Request request;
private volatile Response response;
}
1. consumer发送请求，阻塞等待响应。
调用代码：
HelloService helloService = (HelloService) appCtx.getBean("hello");
String hello = helloService.sayHello();
调用代理对象的sayHello方法，代理类proxy0由javassist动态生成，代码大致如下：
class com.alibaba.dubbo.common.bytecode.proxy0 implements
com.alibaba.dubbo.rpc.service.EchoService, com.zhang.HelloService {
public <init>(java.lang.reflect.InvocationHandler arg0){
handler=$1;
}
public static java.lang.reflect.Method[] methods;
private java.lang.reflect.InvocationHandler handler; public java.lang.String sayHello(){
Object[] args = new Object[0];
//handler是InvokerInvocationHandler对象
Object ret = handler.invoke(this, methods[0], args);
return (java.lang.String)ret;
}
public java.lang.Object $echo(java.lang.Object arg0){
Object[] args = new Object[1];
args[0] = ($w)$1;
Object ret = handler.invoke(this, methods[1], args);
return (java.lang.Object)ret;
}
}
进InvokerInvocationHandler
//InvokerInvocationHandler类
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
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
return invoker.invoke(new RpcInvocation(method, args)).recreate();
}
跳过中间步骤，进到DubboInvoker的doInvoke方法，只分析一般的阻塞调用
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) {
ResponseFuture future = currentClient.request(inv, timeout) ;
RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
return new RpcResult();
} else {
//currentClient对象是 ReferenceCountExchangeClient/HeaderExchangeClient/HeaderExchangeChannel
//拆分来看：ResponseFuture responseFuture = currentClient.request(inv, timeout);
// responseFuture.get(); 调用get导致调用线程阻塞
RpcContext.getContext().setFuture(null);
return (Result) currentClient.request(inv, timeout).get();
}
} catch (TimeoutException e) {
throw new RpcException(RpcException.TIMEOUT_EXCEPTION,
"Invoke remote method timeout. method: " + invocation.getMethodName() +
", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
} catch (RemotingException e) {
throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " +
invocation.getMethodName() + ", provider: " + getUrl() +
", cause: " + e.getMessage(), e);
}
}
进HeaderExchangeChannel
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("2.0.0");
req.setTwoWay(true);
req.setData(request);
DefaultFuture future = new DefaultFuture(channel, req, timeout);
try{
channel.send(req);
}catch (RemotingException e) {
future.cancel();
throw e;
}
return future;
}
进DefaultFuture
public Object get() throws RemotingException {
return get(timeout);
}
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()) {
//调用线程等待直到超时，或者被DubboClientHandler线程唤醒
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());
}
2. 在DubboClientHandler线程中，consumer接收响应，唤醒调用线程。
从ChannelEventRunnable进
//ChannelEventRunnable类
public void run() {
//省略其他代码
switch (state) {
case RECEIVED:
handler.received(channel, message);break;
default:
logger.warn("unknown state: " + state + ", message is " + message);
}
}
进DecodeHandler
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());
}
handler.received(channel, message);
}
进HeaderExchangeHandler
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) {
// handle 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) {
if (isClientSide(channel)) {
Exception e = new Exception("Dubbo client can not supported string message: "
+ message + " in channel: " + channel + ", url: " + channel.getUrl());
logger.error(e.getMessage(), e);
} 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);
}
}
static void handleResponse(Channel channel, Response response) throws RemotingException {
if (response != null && !response.isHeartbeat()) {
DefaultFuture.received(channel, response);
}
}
进DefaultFuture
public static void received(Channel channel, Response response) {
try {
//收到响应删除future
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 {
//给响应赋值
response = res;
if (done != null) {
//DubboClientHandler线程唤醒调用线程
done.signal();
}
} finally {
lock.unlock();
}
if (callback != null) {
invokeCallback(callback);
}
}
可以看到实际是两个线程在通信，main线程发送调用请求，并阻塞，DubboClientHandler线程接收到响应，并唤醒主线程。
DubboClientHandler是一个线程池，它执行工作队列中的任务（即ChannelEventRunnable对象）。
又是谁把ChannelEventRunnable对象放到DubboClientHandler线程池的工作队列中的呢？答案是New I/O client worker
//AllChannelHandler类
public void received(Channel channel, Object message) throws RemotingException {
//DubboClientHandler线程池
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 sayHello();
但配置不是oneway，那么cosumer还是会返回RpcResult，但是RpcResult中的内容为空。