深入理解Dubbo-7.服务消费调用源码分析（下）-阿里云开发者社区

深入理解Dubbo-7.服务消费调用源码分析（中）:https://developer.aliyun.com/article/1414100

handler.channelRead()

服务端收到读的请求是，会进入这个方法。

接着通过handler.received来处理msg，这个handle的链路很长，比较复杂，我们需要逐步剖析

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        NettyChannel channel = NettyChannel.getOrAddChannel(ctx.channel(), this.url, this.handler);
        this.handler.received(channel, msg);
    }

服务端收到读的请求是，会进入这个方法。接着通过handler.received来处理msg ，而这个handler 是在服务发布的时候构建得。

DubboProtocol.createServer:

server = Exchangers.bind(url, this.requestHandler);

Exchanger.bind

public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        if (url == null) {
            throw new IllegalArgumentException("url == null");
        } else if (handler == null) {
            throw new IllegalArgumentException("handler == null");
        } else {
            url = url.addParameterIfAbsent("codec", "exchange");
            return getExchanger(url).bind(url, handler);
        }
    }

通过扩展点选择到HeaderExchanger

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
        return new HeaderExchangeServer(Transporters.bind(url, new ChannelHandler[]{new DecodeHandler(new HeaderExchangeHandler(handler))}));
    }

起链路如下：

MultiMessageHandler(复合消息处理) —> HeartbeatHandle(心跳消息处理，接收心跳并发送心跳响应) —> AllChannelHandler (业务线程转化处理器，把接收到的消息封装成ChannelEventRunnable可执行任

务给线程池处理)—> DecodeHandler (业务解码处理器)—> HeaderExchangeHandler —> DubboProtocol#requestHandler(new ExchangeHandlerAdapter())

而在构建 NettyServerHandler 得时候将 this 传了进去。this 即 NettyServer 。NettyServer是 AbstractPeer 得子类。所以 handler.received 此时会调用AbsstractPeer.received方法，这个方法用来判断服务端是否关闭了，如果关闭就直接返回，否则，通过handler处理链进行层层调用。

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

HeaderExchangeHandler.received

交互层请求响应处理，有三种处理方式

handlerRequest，双向请求
handler.received 单向请求
handleResponse 响应消息

public void received(Channel channel, Object message) throws RemotingException {
        ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
      // 如果接收到的消息是请求类型    
      if (message instanceof Request) {
            Request request = (Request)message;
             // 如果请求是事件类型
            if (request.isEvent()) {
                this.handlerEvent(channel, request);
            // 如果请求是双向类型
            } else if (request.isTwoWay()) {
                this.handleRequest(exchangeChannel, request);
            // 如果请求不是事件类型也不是双向类型
            } else {
                this.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 = this.handler.telnet(channel, (String)message);
                if (echo != null && echo.length() > 0) {
                    channel.send(echo);
                }
            }
          // 如果接收到的消息不是请求、响应或字符串类型
        } else {
            this.handler.received(exchangeChannel, message);//单向请求
        }
    }

handleRequest

接着调用handleRequest方法。这个方法中，构建返回的对象Response，并且最终会通过异步的方式来把msg传递到invoker中进行调用 handler.reply

void handleRequest(final ExchangeChannel channel, Request req) throws RemotingException {
        Response res = new Response(req.getId(), req.getVersion());
        Object data;
        if (req.isBroken()) {
            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((byte)40);
            channel.send(res);
        } else {
            data = req.getData();
            try {
                CompletionStage<Object> future = this.handler.reply(channel, data);// 可以返回一个结果
                future.whenComplete((appResult, t) -> {
                    try {
                        if (t == null) {
                            res.setStatus((byte)20);
                            res.setResult(appResult);
                        } else {
                            res.setStatus((byte)70);
                            res.setErrorMessage(StringUtils.toString(t));
                        }
                        channel.send(res);
                    } catch (RemotingException var5) {
                        logger.warn("Send result to consumer failed, channel is " + channel + ", msg is " + var5);
                    }
                });
            } catch (Throwable var6) {
                res.setStatus((byte)70);
                res.setErrorMessage(StringUtils.toString(var6));
                channel.send(res);
            }
        }
    }

此时的handler.reply，应该是DubboProtocol中构建的匿名内部类

所以调用handler.reply方法，自然就进入到了该匿名内部类中的reply方法中来。

DubboProtocol$requestHandler

private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
        @Override
        public CompletableFuture<Object> reply(ExchangeChannel channel, Object message) throws RemotingException {
　　　　　　  //如果消息类型不是invocation，则抛出异常表示无法识别
            if (!(message instanceof Invocation)) {
                throw new RemotingException(channel, "Unsupported request: "
                        + (message == null ? null : (message.getClass().getName() + ": " + message))
                        + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
            }
　　　　　　  //获得请求参数
            Invocation inv = (Invocation) message;
　　　　      // 获取 invoker 领域对象，这个对象是在发布服务的时候构建，然后封装成 exporter 存在map里面的。
            //根据key从发布的服务列表中查找到指定的服务端invoke，这个就是之前在讲服务发布时，涉及到的invoke对象。
            Invoker<?> invoker = getInvoker(channel, inv);
            // need to consider backward-compatibility if it's a callback
            if (Boolean.TRUE.toString().equals(inv.getObjectAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                String methodsStr = invoker.getUrl().getParameters().get("methods");
                boolean hasMethod = false;
                if (methodsStr == null || !methodsStr.contains(",")) {
                    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());
            //发起请求调用,此时得到的invoker对象
            Result result = invoker.invoke(inv); // 发起对应调用
            return result.thenApply(Function.identity());
        }
　　//......省略代码
};

getInvoker

Invoker<?> invoker = getInvoker(channel, inv);

相当于根据key来获取一个value值

回顾下之前，在发布的时候，调用了一个DubboProtocol.export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter(invoker, key, this.exporterMap);
      // 构建好了之后，把key 和 value存进去
        this.exporterMap.put(key, exporter);
    }
// 而getInvoker也会从map中拿到这个值
Invoker<?> getInvoker(Channel channel, Invocation inv) throws RemotingException {
        ......
        DubboExporter<?> exporter = (DubboExporter)this.exporterMap.get(serviceKey);
        if (exporter == null) {
            throw new RemotingException(channel, "Not found exported service: " + serviceKey + " in " + this.exporterMap.keySet() + ", may be version or group mismatch , channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress() + ", message:" + this.getInvocationWithoutData(inv));
        } else {
            return exporter.getInvoker();
        }
    }

invoker.invoke()

invoker.invoke，发起本地服务调用，但是此时调用之前，invoke并不是一个直接调用的对象，而是包装过的。

在 ServiceConfig#doExportUrlsFor1Protocol 构建包装。最后的调用链路如下：

RegistryProtocol.InvokerDelegate.invoke —> DelegateProviderMetaDataInvoker.invoke —> AbstractProxyInvoker.invoke —> AbstractProxyInvoker(JavassistProxyFactory#getInvoker)

InvokerDelegate 未实现父类 InvokerWrapper invoke方法。进入到InvokerWrapper.invoke方法，这个是一个Invoker包装类，包装了URL地址信息和真正的Invoker代理对象。

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

DelegateProviderMetaDataInvoker

这里是一个委派类，它提供了服务提供者的元数序信息。

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

AbstractProxyInvoker

接着进入到AbstractProxyInvoker的invoke方法，在这个方法中，我们可以看到它会调用子类的doInvoke方法，获得返回结果。

其中proxy，表示服务端的对象实例，这个实例很显然是在构建动态代理Invoker对象时保存进来的。

public Result invoke(Invocation invocation) throws RpcException {
        try {
            Object value = this.doInvoke(this.proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments());
            CompletableFuture<Object> future = this.wrapWithFuture(value);
            CompletableFuture<AppResponse> appResponseFuture = future.handle((obj, t) -> {
                AppResponse result = new AppResponse();
                if (t != null) {
                    if (t instanceof CompletionException) {
                        result.setException(t.getCause());
                    } else {
                        result.setException(t);
                    }
                } else {
                    result.setValue(obj);
                }
                return result;
            });
            return new AsyncRpcResult(appResponseFuture, invocation);
        } catch (InvocationTargetException var5) {
            if (RpcContext.getContext().isAsyncStarted() && !RpcContext.getContext().stopAsync()) {
                this.logger.error("Provider async started, but got an exception from the original method, cannot write the exception back to consumer because an async result may have returned the new thread.", var5);
            }
            return AsyncRpcResult.newDefaultAsyncResult((Object)null, var5.getTargetException(), invocation);
        } catch (Throwable var6) {
            throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + this.getUrl() + ", cause: " + var6.getMessage(), var6);
        }
    }

JavassistProxyFactory.doInvoke

最后进入到具体的子类，也就是在服务的发布的时候通过构建的

@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
        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);
            }
        };
}

服务实例是什么时候生成的

从上面的代码中可以看到，getInvoker中传递的proxy，实际就是对象实例，而这个参数是在serviceConfig中，

Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);

而 ref这个成员变量，是在spring启动时创建bean对象时，会注入这个对象的实例保存到ref中。

总结

至此，服务消费的处理流程就分析完了。

ow new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + this.getUrl() + ", cause: " + var6.getMessage(), var6);

}

#### JavassistProxyFactory.doInvoke
最后进入到具体的子类，也就是在服务的发布的时候通过 构建的
```java
@Override
public <T> Invoker<T> getInvoker(T proxy, Class<T> type, URL url) {
        // TODO Wrapper cannot handle this scenario correctly: the classname contains '$'
        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);
            }
        };
}

服务实例是什么时候生成的

从上面的代码中可以看到，getInvoker中传递的proxy，实际就是对象实例，而这个参数是在serviceConfig中，

Invoker<?> invoker = PROXY_FACTORY.getInvoker(ref, (Class) interfaceClass, url);

而 ref这个成员变量，是在spring启动时创建bean对象时，会注入这个对象的实例保存到ref中。

总结

至此，服务消费的处理流程就分析完了。

深入理解Dubbo-7.服务消费调用源码分析（下）

handler.channelRead()

DubboProtocol.createServer:

Exchanger.bind

通过扩展点选择到HeaderExchanger

HeaderExchangeHandler.received

handleRequest

DubboProtocol$requestHandler

getInvoker

invoker.invoke()

DelegateProviderMetaDataInvoker

AbstractProxyInvoker

JavassistProxyFactory.doInvoke

服务实例是什么时候生成的

总结

服务实例是什么时候生成的

总结

热门文章

最新文章

相关课程

相关电子书

相关实验场景