Dubbo源碼分析（五）Dubbo調用鏈-服務端

• Dubbo源碼分析（一）Dubbo的擴展點機制
• Dubbo源碼分析（二）Dubbo服務發布Export
• Dubbo源碼分析（三）Dubbo的服務引用Refer
• Dubbo源碼分析（四）Dubbo調用鏈-消費端（集群容錯機制）
• Dubbo源碼分析（五）Dubbo調用鏈-服務端
• Dubbo源碼分析（六）Dubbo通信的編碼解碼機制
• Dubbo框架的設計細節 (未完，待續)

這篇來分析Dubbo服務端接收消費端請求調用的過程，先看一張調用鏈的整體流程圖

上面綠色的部分為服務端接收請求部分，大體流程是從ThreadPool-->server-->Exporter-->Filter-->Invoker-->Impl，下面來看源碼

源碼入口

我們知道Dubbo默認是通過Netty進行網絡傳輸，所以這里的源碼入口我們應該找到NettyHandler的接收消息的方法

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

這里就是服務端接收消費端發送請求的地方，進入handler.received方法，在AbstractPeer類中

public void received(Channel ch, Object msg) throws RemotingException {
        if (closed) {
            return;
        }
        handler.received(ch, msg);
    }

進入到handler.received,最終我們進入AllChannelHandler.received方法中

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方法，我們看一下

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,先進行解碼，這里的內容暫時不說，放在后面的解編碼一起說，繼續進入HeaderExchangeHandler.received方法，

public void received(Channel channel, Object message) throws RemotingException {
        ···
                        Response response = handleRequest(exchangeChannel, request);
                        channel.send(response);
                    } else {
                        handler.received(exchangeChannel, request.getData());
                    }
                }
            } ···
        } finally {
            HeaderExchangeChannel.removeChannelIfDisconnected(channel);
        }
    }

執行handleRequest(exchangeChannel, request)方法，這里是網絡通信接收處理方法，繼續走，進入DubboProtocol.reply方法

public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
            if (message instanceof Invocation) {
                Invocation inv = (Invocation) message;
                Invoker<?> invoker = getInvoker(channel, inv);
                ···
                return invoker.invoke(inv);
            }
           ···
        }

進入getInvoker方法，

 Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
       ···
        DubboExporter<?> exporter = (DubboExporter<?>) exporterMap.get(serviceKey);
       ···
        return exporter.getInvoker();
    }

從exporterMap中獲取所需的exporter，還記不記得這個exporterMap是什么時候放入值的，在服務端暴露接口的時候，這個是在第二篇中有提到過

然后執行exporter.getInvoker()，現在我們要將exporter轉化為invoeker對象，我們拿到這個invoker對象，並執行invoker.invoke(inv)方法，然后經過8個Filter，最終進入JavassistProxyFactory.AbstractProxyInvoker.doInvoke方法，

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

還記得這段代碼嗎？這個wrapper是什么？下面這個截圖是在服務暴露的時候，

這個時候執行 wrapper.invokeMethod方法，此時的wrapper就是之前我們動態生成的一個wrapper包裝類，然后進入真正的實現類DemoServiceImpl.sayHello方法，執行完成之后，回到HeaderExchangeHandler.received方法

channel.send(response)

最終把結果通過ChannelFuture future = channel.write(message)發回consumer

消費端接收服務端發送的執行結果

我們先進入NettyHandler.messageReceived方法，再執行handler.received(channel, e.getMessage())，這個和上面的代碼是一樣的，繼續進入MultiMessageHandler.received方法，繼續進入ChannelEventRunnable線程池，繼續進入DecodeHandler.received解碼，最終進入DefaultFuture.doReceived方法

private void doReceived(Response res) {
        lock.lock();
        try {
            response = res;
            if (done != null) {
                done.signal();
            }
        } finally {
            lock.unlock();
        }
        if (callback != null) {
            invokeCallback(callback);
        }
    }

這里用到了一個Condition，就是這個done，這里的done.signal的作用是什么呢？既然有signal方法，一定還有done.await方法,我們看到這個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();
    }

這個里面有一個done.await方法，貌似這個get方法有點熟悉，在哪里見過呢，在DubboInvoker.doInvoke()方法中的最后一行,貼代碼

return (Result) currentClient.request(inv, timeout).get();

這里拿到消費端發起請求之后，調用了get方法，這個get方法一直阻塞，直到服務端返回結果調用done.signal()，這個也是Dubbo的異步轉同步機制的實現方式，至此，Dubbo的調用鏈就分析完了