07.Dubbo源码系列V1-Dubbo第七节-服务调用源码解析

• • 第七节: Dubbo服务调用源码解析
  • • 服务导出的Netty启动源码
    • • DubboProtocol
      • Exchangers
      • HeaderExchange
      • HeaderExchangeServer
      • Transporters
      • NettyTransporter
      • NettyServer
      • ChannelHandlers
      • AbstractServer
      • NettyServer
      • 分割
    • 服务提供端执行逻辑
    • • 概述
      • JavassistProxyFactory
    • 服务消费端执行逻辑
    • Dubbo的异常处理

1. 文章可能会优先更新在Github，包括文章纠错与增加内容，其它平台会晚一段时间。
2. 转载须知：转载请注明GitHub出处，让我们一起维护一个良好的技术创作环境。
3. 如果你要提交 issue 或者 pr 的话建议到 Github 提交。笔者会陆续更新，如果对你有所帮助，不妨Github点个Star~。你的Star是我创作的动力。

最主要的就是构造一个Handler处理链路

    public  Exporter export(Invoker invoker) throws RpcException {
        URL url = invoker.getUrl();

        // 唯一标识一个服务的key
        String key = serviceKey(url);
        // 构造一个Exporter进行服务导出
        DubboExporter exporter = new DubboExporter(invoker, key, exporterMap);
        exporterMap.put(key, exporter);

        // 省略...

        // 开启NettyServer
        // 请求--->invocation--->服务key--->exporterMap.get(key)--->exporter--->invoker--->invoker.invoke(invocation)-->执行服务
        openServer(url);

        return exporter;
    }

    private void openServer(URL url) {
        // find server.
        String key = url.getAddress(); // 获得ip地址和port， 192.168.40.17:20880

        // NettyClient, NettyServer
        //client can export a service which's only for server to invoke
        boolean isServer = url.getParameter(IS_SERVER_KEY, true);
        if (isServer) {
            // 缓存Server对象
            ExchangeServer server = serverMap.get(key);

            // DCL，Double Check Lock
            if (server == null) {
                synchronized (this) {
                    server = serverMap.get(key);
                    if (server == null) {
                        // 创建Server，并进行缓存
                        serverMap.put(key, createServer(url));
                    }
                }
            } else {
                // server supports reset, use together with override
                // 服务重新导出时，就会走这里
                server.reset(url);
            }
        }
    }

    private ExchangeServer createServer(URL url) {
        url = URLBuilder.from(url)
                // send readonly event when server closes, it's enabled by default
                .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
                // enable heartbeat by default
                .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
                .addParameter(CODEC_KEY, DubboCodec.NAME)
                .build();

        // 协议的服务器端实现类型，比如：dubbo协议的mina,netty等，http协议的jetty,servlet等，默认为netty
        String str = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);

        if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);
        }

        // 通过url绑定端口，和对应的请求处理器
        ExchangeServer server;
        try {
            // requestHandler是请求处理器，类型为ExchangeHandler
            // 表示从url的端口接收到请求后，requestHandler来进行处理
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) {
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }

        // 协议的客户端实现类型，比如：dubbo协议的mina,netty等
        str = url.getParameter(CLIENT_KEY);
        if (str != null && str.length() > 0) {
            Set supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) {
                throw new RpcException("Unsupported client type: " + str);
            }
        }

        return server;
    }

1. NettyClient<------>Socket连接，数据传输层<------>NettyServer。Netty这两端只要建立了连接就可以互相发送数据。
2. ExchangeClient------数据交换层------ExchangeServer。这是Dubbo抽象出来的概念，主要就是抽象出了请求和响应这两个概念。
3. ExchangeXXX里面包了Netty的东西

private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

    @Override
    public CompletableFuture