dubbo源码解析（四十七）服务端处理请求过程

摘要：而存在的意义就是保证请求或响应对象可在线程池中被解码，解码完成后，就会分发到的。

目标：从源码的角度分析服务端接收到请求后的一系列操作，最终把客户端需要的值返回。

上一篇讲到了消费端发送请求的过程，该篇就要将服务端处理请求的过程。也就是当服务端收到请求数据包后的一系列处理以及如何返回最终结果。我们也知道消费端在发送请求的时候已经做了编码，所以我们也需要在服务端接收到数据包后，对协议头和协议体进行解码。不过本篇不讲解如何解码。有兴趣的可以翻翻我以前的文章，有讲到关于解码的逻辑。接下来就开始讲解服务端收到请求后的逻辑。

假设远程通信的实现还是用netty4，解码器将数据包解析成 Request 对象后，NettyHandler 的 messageReceived 方法紧接着会收到这个对象，所以第一步就是NettyServerHandler的channelRead。

可以参考《dubbo源码解析（十七）远程通信——Netty4》的（三）NettyServerHandler

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    // 看是否在缓存中命中，如果没有命中，则创建NettyChannel并且缓存。
    NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), url, handler);
    try {
        // 接受消息
        handler.received(channel, msg);
    } finally {
        // 如果通道不活跃或者断掉，则从缓存中清除
        NettyChannel.removeChannelIfDisconnected(ctx.channel());
    }
}

NettyServerHandler是基于netty4实现的服务端通道处理实现类，而该方法就是用来接收请求，接下来就是执行AbstractPeer的received。

可以参考《dubbo源码解析（九）远程通信——Transport层》的（一）AbstractPeer

public void received(Channel ch, Object msg) throws RemotingException {
    // 如果通道已经关闭，则直接返回
    if (closed) {
        return;
    }
    handler.received(ch, msg);
}

该方法比较简单，之前也讲过AbstractPeer类就做了装饰模式中装饰角色，只是维护了通道的正在关闭和关闭完成两个状态。然后到了MultiMessageHandler的received

可以参考《dubbo源码解析（九）远程通信——Transport层》的（八）MultiMessageHandler

public void received(Channel channel, Object message) throws RemotingException {
    // 如果消息是MultiMessage类型的，也就是多消息类型
    if (message instanceof MultiMessage) {
        // 强制转化为MultiMessage
        MultiMessage list = (MultiMessage) message;
        // 把各个消息进行发送
        for (Object obj : list) {
            handler.received(channel, obj);
        }
    } else {
        // 直接发送
        handler.received(channel, message);
    }
}

该方法也比较简单，就是对于多消息的处理。

可以参考《dubbo源码解析（十）远程通信——Exchange层》的（二十）HeartbeatHandler。其中就是对心跳事件做了处理。如果不是心跳请求，那么接下去走到AllChannelHandler的received。

可以参考《dubbo源码解析（九）远程通信——Transport层》的（十一）AllChannelHandler。该类处理的是连接、断开连接、捕获异常以及接收到的所有消息都分发到线程池。所以这里的received方法就是把请求分发到线程池，让线程池去执行该请求。

还记得我在之前文章里面讲到到Dispatcher接口吗，它是一个线程派发器。分别有五个实现：

这些Dispatcher的实现类以及对应的Handler都可以在《dubbo源码解析（九）远程通信——Transport层》中查看相关实现。dubbo默认all为派发策略。所以我在这里讲了AllChannelHandler的received。把消息送到线程池后，可以看到首先会创建一个ChannelEventRunnable实体。那么接下来就是线程接收并且执行任务了。

ChannelEventRunnable实现了Runnable接口，主要是用来接收消息事件，并且根据事件的种类来分别执行不同的操作。来看看它的run方法：

public void run() {
    // 如果是接收的消息
    if (state == ChannelState.RECEIVED) {
        try {
            // 直接调用下一个received
            handler.received(channel, message);
        } catch (Exception e) {
            logger.warn("ChannelEventRunnable handle " + state + " operation error, channel is " + channel
                    + ", message is " + message, e);
        }
    } else {
        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 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，所以多带带先做处理，根据不同的类型的消息，会被执行不同的逻辑，我们这里主要看state为RECEIVED的，那么如果是RECEIVED，则会执行下一个received方法。

可以参考《dubbo源码解析（九）远程通信——Transport层》的（七）DecodeHandler。可以看到received方法中根据消息的类型进行不同的解码。而DecodeHandler 存在的意义就是保证请求或响应对象可在线程池中被解码，解码完成后，就会分发到HeaderExchangeHandler的received。

public void received(Channel channel, Object message) throws RemotingException {
    // 设置接收到消息的时间戳
    channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis());
    // 获得通道
    final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
    try {
        // 如果消息是Request类型
        if (message instanceof Request) {
            // handle request.
            // 强制转化为Request
            Request request = (Request) message;
            // 如果该请求是事件心跳事件或者只读事件
            if (request.isEvent()) {
                // 执行事件
                handlerEvent(channel, request);
            } else {
                // 如果是正常的调用请求，且需要响应
                if (request.isTwoWay()) {
                    // 处理请求
                    handleRequest(exchangeChannel, request);
                } else {
                    // 如果不需要响应，则继续下一步
                    handler.received(exchangeChannel, request.getData());
                }
            }
        } else if (message instanceof Response) {
            // 处理响应
            handleResponse(channel, (Response) message);
        } else if (message instanceof String) {
            // 如果是telnet相关的请求
            if (isClientSide(channel)) {
                // 如果是客户端侧，则直接抛出异常，因为客户端侧不支持telnet
                Exception e = new Exception("Dubbo client can not supported string message: " + message + " in channel: " + channel + ", url: " + channel.getUrl());
                logger.error(e.getMessage(), e);
            } 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);
    }
}

该方法中就对消息进行了细分，事件请求、正常的调用请求、响应、telnet命令请求等，并且针对不同的消息类型做了不同的逻辑调用。我们这里主要看正常的调用请求。见下一步。

void handleRequest(final ExchangeChannel channel, Request req) throws RemotingException {
    // 创建一个Response实例
    Response res = new Response(req.getId(), req.getVersion());
    // 如果请求被破坏了
    if (req.isBroken()) {
        // 获得请求的数据包
        Object data = req.getData();

        String msg;
        // 如果数据为空
        if (data == null) {
            //消息设置为空
            msg = null;
            // 如果在这之前已经出现异常，也就是数据为Throwable类型
        } 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);

        // 发送该消息
        channel.send(res);
        return;
    }
    // find handler by message class.
    // 获得请求数据 也就是 RpcInvocation 对象
    Object msg = req.getData();
    try {
        // 继续向下调用 返回一个future
        CompletionStage