Dubbo消费者调用过程源码解析

发表于 | 分类于 | 阅读次数：

前言

之前已经写了三篇关于DUBBO的了，本文将是DUBBO系列的最后最后一篇，这篇写完之后也就对DUBBO的整个初始化和调用流程都有了较为详细的了解了，让我们赶紧来做个Ending吧。

源码解析

消费者调用流程涉及到消费者端和生产者端的交互，所以将分为三个部分来讲解，分别是
-消费者发起调用请求
-生产者响应调用请求
-消费者获取调用结果

消费者发起调用请求

之前文章中讲过消费者初始化时最后返回的是一个InvokerInvocationHandler的代理对象，根据动态代理的原理，DUBBO接口的方法调用都会由invoke方法代理，我们来看一下其实现

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
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]);
    }
    return invoker.invoke(new RpcInvocation(method, args)).recreate();
}

正常情况下的方法调用会走invoker.invoke(new RpcInvocation(method, args)).recreate()这个分支，首先来看new RpcInvocation(method, args)

1
2
3
4
5
6
7
8
9
10
11
public RpcInvocation(Method method, Object[] arguments) {
    this(method.getName(), method.getParameterTypes(), arguments, null, null);
}

public RpcInvocation(String methodName, Class<?>[] parameterTypes, Object[] arguments, Map<String, String> attachments, Invoker<?> invoker) {
    this.methodName = methodName;
    this.parameterTypes = parameterTypes == null ? new Class<?>[0] : parameterTypes;
    this.arguments = arguments == null ? new Object[0] : arguments;
    this.attachments = attachments == null ? new HashMap<String, String>() : attachments;
    this.invoker = invoker;
}

非常简单的一个初始化赋值操作，就不做过多讲解了，接着回头看invoker.invoke(new RpcInvocation(method, args))方法，这里的invoker之前也说过了，是一个通过SPI机制生成的对象，以默认设置的参数failover为例，这里的invoker就是一个MockClusterInvoker对象中包含了一个FailoverClusterInvoker对象引用的类似链式的对象，那么我们来详细看看MockClusterInvoker的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
28
29
30
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")){
       	//no mock
       	result = this.invoker.invoke(invocation);
       } else if (value.startsWith("force")) {
       	if (logger.isWarnEnabled()) {
       		logger.info("force-mock: " + invocation.getMethodName() + " force-mock enabled , url : " +  directory.getUrl());
       	}
       	//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 {
				if (logger.isWarnEnabled()) {
	        		logger.info("fail-mock: " + invocation.getMethodName() + " fail-mock enabled , url : " +  directory.getUrl(), e);
	        	}
				result = doMockInvoke(invocation, e);
			}
		}
       }
       return result;
}

当没有配置mock值时，value值得到的是默认值false，会去执行result = this.invoker.invoke(invocation)，this.invoker刚才提到过了是一个FailoverClusterInvoker类型的对象，但该对象并没有实现invoke方法，实际上该方法是继承自父类AbstractClusterInvoker的，来看一下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
public Result invoke(final Invocation invocation) throws RpcException {

    checkWheatherDestoried();

    LoadBalance loadbalance;
    
    List<Invoker<T>> invokers = list(invocation);
    if (invokers != null && invokers.size() > 0) {
        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);
    return doInvoke(invocation, invokers, loadbalance);
}

这里的list(invocation)方法根据invocation中的参数来获取所有的invoker列表，就不深入讲了，接着来看loadbalance对象的生成，loadbalance对象根据SPI机制生成，具体实现由loadbalance参数决定，也就是具体的负载均衡策略，DUBBO提供的实现有random、roundrobin、leastactive、consistenthash四种，其中没有根据服务端负载进行调节的策略。其中默认实现为random，生成的loadbalance就是一个RandomLoadBalance的对象。本次只分析同步的接口调用方式，跳过RpcUtils.attachInvocationIdIfAsync，接着看doInvoke(invocation, invokers, loadbalance)方法，该方法实现在FailoverClusterInvoker中

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
40
41
42
43
44
45
46
47
48
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.
    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();
                    //获得InvokerWrapper的List
    		copyinvokers = list(invocation);
    		//重新检查一下
    		checkInvokers(copyinvokers, invocation);
    	}
        Invoker<T> invoker = select(loadbalance, invocation, copyinvokers, invoked);
        invoked.add(invoker);
        RpcContext.getContext().setInvokers((List)invoked);
        try {
            Result result = invoker.invoke(invocation);
            if (le != null && logger.isWarnEnabled()) {
                logger.warn("");
            }
            return result;
        } catch (RpcException e) {
            if (e.isBiz()) { // biz exception.
                throw e;
            }
            le = e;
        } catch (Throwable e) {
            le = new RpcException(e.getMessage(), e);
        } finally {
            providers.add(invoker.getUrl().getAddress());
        }
    }
    throw new RpcException
     /**
     *   略去部分代码
     */
}

这里select(loadbalance, invocation, copyinvokers, invoked)方法根据传入的loadbalance对象挑选出一个执行用的invoker，里面调用链较深，在此不做详细分析。最终将通过invoker.invoke(invocation)进行调用并返回一个Result类型的对象，也就是最终的执行结果，这里的invoker对象是InvokerWrapper的实例，该实例引用了一个ListenerInvokerWrapper的实例，接着又链式引用了AbstractInvoker的实例，因此最终执行的invoke方法在AbstractInvoker中，来看一下

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
public Result invoke(Invocation inv) throws RpcException {
    if(destroyed) {
        throw new RpcException("略");
    }
    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());
    }
    //异步操作默认添加invocation id
    RpcUtils.attachInvocationIdIfAsync(getUrl(), invocation);
    
    
    try {
        return doInvoke(invocation);
    } catch (InvocationTargetException e) { // biz exception
     /**
     *   略去部分代码
     */
    }
}

这里的关键方法是doInvoke(invocation)，其实现在具体的Invoker实现类中，这里我们采用的是默认的dubbo协议，所以实现类为DubboInvoker，来看看其doInvoke方法

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
@Override
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 {
        	RpcContext.getContext().setFuture(null);
            return (Result) currentClient.request(inv, timeout).get();
        }
     /**
     *   略去部分代码
     */
}

这里的isOneway和isAsync两个标志位分别区分单向调用(不在乎调用结果)和异步调用，这里我们分析同步调用的流程，这里的currentClient是一个ReferenceCountExchangeClient类型的对象

1
2
3
public ResponseFuture request(Object request) throws RemotingException {
    return client.request(request);
}

这里的client是一个HeaderExchangeClient类型的对象，

1
2
3
public ResponseFuture request(Object request) throws RemotingException {
    return channel.request(request);
}

这里的channel是一个HeaderExchangeChannel类型的对象，继续跟进去

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
public ResponseFuture request(Object request) throws RemotingException {
    return request(request, channel.getUrl().getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
}

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;
}

这里的request方法自己又进行了一次内部调用，可以看到具体实现时创建了一个DefaultFuture对象并且通过channel.send(req)方法发送请求到生产者端，这里不做具体深入了。接着我们跳回DubboInvoker类doInvoke方法中的currentClient.request(inv, timeout).get()，这里是不是和jdk中future的用法很像，事实上这里也确实是通过get方法的调用将线程阻塞在这里等待结果，从而将异步调用转化为同步。为了证实这个想法，我们来看看DefaultFuture的get方法

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
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()) {
                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();
}

从done.await(timeout, TimeUnit.MILLISECONDS)可以看到这里不仅是等待isDone()这个状态位，同时还有超时时间的限制。isDone()判断的是什么，来看一下

1
2
3
public boolean isDone() {
    return response != null;
}

判断response对象是否为空，那么后面的流程其实不难猜，生产者处理完结果会来填充response。

生产者响应调用请求

生产者开启了端口监听，消息的解码由Netty处理，解码后交由NettyHandler的messageReceived方法进行业务处理，来看一下

1
2
3
4
5
6
7
8
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
    try {
        handler.received(channel, e.getMessage());
    } finally {
        NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
    }
}

先来看一下NettyChannel.getOrAddChannel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
static NettyChannel getOrAddChannel(org.jboss.netty.channel.Channel ch, URL url, ChannelHandler handler) {
    if (ch == null) {
        return null;
    }
    NettyChannel ret = channelMap.get(ch);
    if (ret == null) {
        NettyChannel nc = new NettyChannel(ch, url, handler);
        if (ch.isConnected()) {
            ret = channelMap.putIfAbsent(ch, nc);
        }
        if (ret == null) {
            ret = nc;
        }
    }
    return ret;
}

主要是从channelMap中获取对应的NettyChannel，接着回到NettyHandler的messageReceived方法来看handler.received(channel, e.getMessage())，这里的handler是一个NettyServer的实例，但它本身没有实现received方法，该方法要追溯到它的父类的父类的父类（真的就是这么长的继承关系。。。）AbstractPeer中，来看一下

1
2
3
4
5
6
public void received(Channel ch, Object msg) throws RemotingException {
    if (closed) {
        return;
    }
    handler.received(ch, msg);
}

这里的handler是MultiMessageHandler对象的实例，来看一下其received方法的实现

1
2
3
4
5
6
7
8
9
10
11
@Override
   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 {
           handler.received(channel, message);
       }
   }

这里的handler又是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
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;
    }
    handler.received(channel, message);
}

因为不是心跳类的消息，所以执行handler.received(channel, message)继续这个调用链，这里的handler是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中到底需要执行什么任务

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
40
41
42
43
44
public void run() {
    switch (state) {
        case CONNECTED:
            try{
                handler.connected(channel);
            }catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
            }
            break;
        case DISCONNECTED:
            try{
                handler.disconnected(channel);
            }catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel, e);
            }
            break;
        case SENT:
            try{
                handler.sent(channel,message);
            }catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                        + ", message is "+ message,e);
            }
            break;
        case RECEIVED:
            try{
                handler.received(channel, message);
            }catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                        + ", message is "+ message,e);
            }
            break;
        case CAUGHT:
            try{
                handler.caught(channel, exception);
            }catch (Exception e) {
                logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is "+ channel
                        + ", message is: " + message + ", exception is " + exception,e);
            }
            break;
        default:
            logger.warn("unknown state: " + state + ", message is " + message);
    }
}

这里传入的是RECEIVED状态，执行对应分支又是调用handler.received(channel, message)，好吧继续。。。
这里的handler是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());
    }

    if (message instanceof Response) {
        decode( ((Response)message).getResult());
    }

    handler.received(channel, message);
}

调用链还在继续，这次的handler是HeaderExchangeHandler类型

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
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);
    }
}

正常同步请求会开始执行handleRequest(exchangeChannel, request)处理请求，并通过channel.send(response)回复结果，来重点看一下handleRequest方法

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
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 {
        // handle data.
        Object result = handler.reply(channel, msg);
        res.setStatus(Response.OK);
        res.setResult(result);
    } catch (Throwable e) {
        res.setStatus(Response.SERVICE_ERROR);
        res.setErrorMessage(StringUtils.toString(e));
    }
    return res;
}

可以看出正常请求将由handler.reply(channel, msg)处理，这里的handler是DubboProtocol中的一个ExchangeHandlerAdapter实现，其reply方法如下

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
public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
    if (message instanceof 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){
                logger.warn(new IllegalStateException("The methodName "+inv.getMethodName()+" not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) +" ,invocation is :"+inv );
                return null;
            }
        }
        RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
        return invoker.invoke(inv);
    }
    throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
}

这里一共做了两件事，先通过getInvoker(channel, inv)获取具体的invoker，再通过invoker.invoke(inv)执行获取结果，先来看一下getInvoker(channel, inv)

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
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));

    DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
    
    if (exporter == null)
        throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + exporterMap.keySet() + ", may be version or group mismatch " + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + inv);

    return exporter.getInvoker();
}

这里又看到了熟悉的exporterMap，之前讲生产者初始化的时候就说过这个map中放入了封装过的Invoker对象exporter，现在又把它取了出了并通过getInvoker()方法获得封装在其中的Invoker对象。
接着来看invoker.invoke(inv)方法，其实现首先在InvokerWrapper类中

1
2
3
public Result invoke(Invocation invocation) throws RpcException {
    return invoker.invoke(invocation);
}

然后会调用到AbstractProxyInvoker中的invoke方法

1
2
3
4
5
6
7
8
9
public Result invoke(Invocation invocation) throws RpcException {
    try {
        return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments()));
    } catch (InvocationTargetException e) {
        return new RpcResult(e.getTargetException());
    } catch (Throwable e) {
        throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

这里doInvoke方法的实现在JavassistProxyFactory中getInvoker方法中

1
2
3
4
5
6
7
8
9
10
11
12
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);
        }
    };
}

这里根据传入的 proxy对象的类信息创建对它的包装对象Wrapper
并调用其invokeMethod方法，通过传入的参数来调用proxy对象的对应方法，返回调用结果，也就是执行具体的业务。
完成handleRequest(exchangeChannel, request)方法的解析后，回到HeaderExchangeHandler类中接着来看一下channel.send(response)，这里的channel传入的是NettyChannel类型的对象，send方法的实现在其父类的父类AbstractPeer中，来看一下

1
2
3
public void send(Object message) throws RemotingException {
    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
public void send(Object message, boolean sent) throws RemotingException {
    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");
    }
}

可以看到业务处理结果最后通过ChannelFuture对象进行了发送，到此生产者端的任务就完成了。

消费者获取调用结果

这里消费者端通过NETTY从生产者端获取数据的流程和之前的如出一辙，调用链直到HeaderExchangeHandler之前都是一样的，我们先来回顾一下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
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);
        }
    }

之前走的是Request分支，这次因为是响应消息走的是Response分支，那么来看一下handleResponse(channel, (Response) message)的具体实现

1
2
3
4
5
static void handleResponse(Channel channel, Response response) throws RemotingException {
    if (response != null && !response.isHeartbeat()) {
        DefaultFuture.received(channel, response);
    }
}

继续跟进去看received方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
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());
    }
}

继续看doReceived干了什么

1
2
3
4
5
6
7
8
9
10
11
12
13
14
private void doReceived(Response res) {
    lock.lock();
    try {
        response = res;
        if (done != null) {
            done.signal();
        }
    } finally {
        lock.unlock();
    }
    if (callback != null) {
        invokeCallback(callback);
    }
}

看到这里把执行结果赋值给response，正好应证了我们之前的猜想，消费者的同步阻塞也就可以继续执行下去了，这也算是非常经典的异步转同步的实现方案了吧。

总结

本文把消费者端和生产者端交互的大概流程进行了讲解，流程主要分为三个部分，分别是：消费者发起调用请求、生产者响应调用请求和消费者获取调用结果，概括一下就是消费者通过生成的代理对象调用invoke方法通过Netty的通道去请求生产者的exporter进行执行，并且通过future的方式将异步的交互转为了同步响应。