[网络协议] Dubbo源码解析-Http协议解析

开发: C++知识库 Java知识库 JavaScript Python PHP知识库人工智能区块链大数据移动开发嵌入式开发工具数据结构与算法开发测试游戏开发网络协议系统运维
教程: HTML教程 CSS教程 JavaScript教程 Go语言教程 JQuery教程 VUE教程 VUE3教程 Bootstrap教程 SQL数据库教程 C语言教程 C++教程 Java教程 Python教程 Python3教程 C#教程
数码: 电脑笔记本显卡显示器固态硬盘硬盘耳机手机 iphone vivo oppo 小米华为单反装机图拉丁

-> 网络协议 -> Dubbo源码解析-Http协议解析 -> 正文阅读

[网络协议]Dubbo源码解析-Http协议解析

前言：

在前文介绍完Dubbo协议的传输之后，我们了解了Dubbo协议主要是定义了head+body，通过head中对每个字节的设置来区分具体的系列化方式，body长度，然后获取对应的body，并反序列化为一个可用的Request对象之后，交由对应的server来处理即可。

本文再来介绍一种协议，一种我们都熟悉的协议：http协议，了解下在该协议下请求是如何传输和响应的。

1.http协议的示例

有关于接口和实现类如之前的示例一样：

public interface DemoService {
    String sayHello(String name);
}

@Service
public class DemoServiceImpl implements DemoService {
    private static final Logger logger = LoggerFactory.getLogger(DemoServiceImpl.class);
    @Override
    public String sayHello(String name) {
        logger.info("Hello " + name + ", request from consumer: " + RpcContext.getContext().getRemoteAddress());
        return "Hello " + name + ", response from provider: " + RpcContext.getContext().getLocalAddress();
    }
}

1.1 provider示例

public class ProviderApplication {
    public static void main(String[] args) throws Exception {
        ServiceConfig<DemoServiceImpl> service = new ServiceConfig<>();
        service.setInterface(DemoService.class);
        service.setRef(new DemoServiceImpl());
        service.setApplication(new ApplicationConfig("dubbo-demo-api-provider"));
        service.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));

        service.setScope("remote");

        // 自定义为http协议，主要是这里有所不同
        ProtocolConfig protocolConfig = new ProtocolConfig();
        protocolConfig.setName("http");
        protocolConfig.setPort(8000);

        service.setProtocol(protocolConfig);
        service.export();

        System.out.println("dubbo service started");
        new CountDownLatch(1).await();
    }
}

相对于Dubbo协议，这里多创建了一个ProtocolConfig对象，用于指定http协议。

1.2 consumer示例

public class ConsumerApplication {
    public static void main(String[] args) {
		ReferenceConfig<DemoService> reference = new ReferenceConfig<>();
        reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
        reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));
        reference.setInterface(DemoService.class);
        reference.setTimeout(10000);
        reference.setScope("remote");

        // 指定请求协议为http协议
        reference.setProtocol("http");

        DemoService service = reference.get();
        String message = service.sayHello("dubbo");
        System.out.println(message);
    }
}

相较于Dubbo协议的消费者，这里就多了一行代码，在Reference中指定协议为http。

读者可以自行测试下，结果与Dubbo协议没有什么不同。

2.HttpProtocol解析

HttpProtocol同样实现了Protocol接口，同样也需要实现export()、refer()接口，分别代表暴露服务、引用服务。

那我们还是从这两个接口出发，分别来分析下基于http协议的服务暴露和服务引用。

2.1 HttpProtocol.export() 暴露服务

public abstract class AbstractProxyProtocol extends AbstractProtocol {
 
    public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
        // 针对http协议而言，当前URL=http://192.168.142.1:8000/org.apache.dubbo.demo.DemoService?anyhost=true&
        // application=dubbo-demo-api-provider&bind.ip=192.168.142.1&bind.port=8000&deprecated=false&dubbo=2.0.2&dynamic=true
        // &generic=false&interface=org.apache.dubbo.demo.DemoService&methods=sayHello&pid=21944&release=
        // &scope=remote&side=provider&timestamp=1637927029141
        
        // 最终获得的uri=org.apache.dubbo.demo.DemoService:8000，得到接口和暴露端口号的组合
        final String uri = serviceKey(invoker.getUrl());
        Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
        if (exporter != null) {
            if (Objects.equals(exporter.getInvoker().getUrl(), invoker.getUrl())) {
                return exporter;
            }
        }
        // 主要还是交由HttpProtocol.doExport()来执行，具体来看2.1.1
        final Runnable runnable = doExport(proxyFactory.getProxy(invoker, true), invoker.getInterface(), invoker.getUrl());
        exporter = new AbstractExporter<T>(invoker) {
            @Override
            public void unexport() {
                super.unexport();
                exporterMap.remove(uri);
                if (runnable != null) {
                    try {
                        runnable.run();
                    } catch (Throwable t) {
                        logger.warn(t.getMessage(), t);
                    }
                }
            }
        };
        exporterMap.put(uri, exporter);
        return exporter;
    }
}

2.1.1 HttpProtocol.doExport() 暴露服务

public class HttpProtocol extends AbstractProxyProtocol {
    private final Map<String, JsonRpcServer> skeletonMap = new ConcurrentHashMap<>();
    private HttpBinder httpBinder;
    
    protected <T> Runnable doExport(final T impl, Class<T> type, URL url) throws RpcException {
        // 本例中addr=0.0.0.0:8000，即服务暴露地址
        String addr = getAddr(url);
        ProtocolServer protocolServer = serverMap.get(addr);
        if (protocolServer == null) {
            // 暴露当前http端口8000
            // 本质上还是交由JettyHttpBinder/TomcatHttpBinder来执行，以应用服务器的方式暴露端口，而请求执行类为InternalHandler
            RemotingServer remotingServer = httpBinder.bind(url, new InternalHandler(url.getParameter("cors", false)));
            // 创建完应用服务器，暴露端口之后，将addr与Server保存到serverMap中
            serverMap.put(addr, new ProxyProtocolServer(remotingServer));
        }
        
        // 获取服务path，本例中为/org.apache.dubbo.demo.DemoService
        final String path = url.getAbsolutePath();
        final String genericPath = path + "/" + GENERIC_KEY;
        // 创建path对应的JsonRpcServer，并保存到skeletonMap，后续处理请求时会用到
        JsonRpcServer skeleton = new JsonRpcServer(impl, type);
        JsonRpcServer genericServer = new JsonRpcServer(impl, GenericService.class);
        skeletonMap.put(path, skeleton);
        skeletonMap.put(genericPath, genericServer);
        return () -> {
            skeletonMap.remove(path);
            skeletonMap.remove(genericPath);
        };
    }
}

针对Http协议而言，同一个服务暴露端口，也是同一个RemotingServer，避免多次创建。

注意：关于JsonRpcServer、JettyHttpBinder相关知识点，笔者不再多做介绍，读者可自行阅读其他文章。

总结：经过JettyHttpBinder的端口暴露以及JsonRpcServer的创建，后续请求可以向对应端口发送http请求。

请求被接收后，将交由InternalHandler来执行，而后最终还是调用JsonRpcServer来完成方法调用，这个后续再说。

2.2 HttpProtocol.refer() 引用服务

最终还是调用HttpProtocol.doRefer()来实现服务引用，我们直接看这个方法的实现

public class HttpProtocol extends AbstractProxyProtocol {
 
    protected <T> T doRefer(final Class<T> serviceType, URL url) throws RpcException {
        // generic相关逻辑，非重点
        final String generic = url.getParameter(GENERIC_KEY);
        final boolean isGeneric = ProtocolUtils.isGeneric(generic) || serviceType.equals(GenericService.class);
        
        // 创建
        JsonProxyFactoryBean jsonProxyFactoryBean = new JsonProxyFactoryBean();
        JsonRpcProxyFactoryBean jsonRpcProxyFactoryBean = new JsonRpcProxyFactoryBean(jsonProxyFactoryBean);
        jsonRpcProxyFactoryBean.setRemoteInvocationFactory((methodInvocation) -> {
            RemoteInvocation invocation = new JsonRemoteInvocation(methodInvocation);
            if (isGeneric) {
                invocation.addAttribute(GENERIC_KEY, generic);
            }
            return invocation;
        });
        // 此时key=http://192.168.142.1:8000/org.apache.dubbo.demo.DemoService
        String key = url.setProtocol("http").toIdentityString();
        if (isGeneric) {
            key = key + "/" + GENERIC_KEY;
        }

        jsonRpcProxyFactoryBean.setServiceUrl(key);
        jsonRpcProxyFactoryBean.setServiceInterface(serviceType);

        // 主要用于创建jsonRpcHttpClient，发起http请求调用
        jsonProxyFactoryBean.afterPropertiesSet();
        return (T) jsonProxyFactoryBean.getObject();
    }
}

最终，还是通过jsonRpcHttpClient来发起调用，如下所示

2.2.1?jsonRpcHttpClient.invoke()

public class JsonRpcHttpClient extends JsonRpcClient implements IJsonRpcClient {
	public Object invoke(String methodName, Object argument, Type returnType, Map<String, String> extraHeaders) throws Throwable {
        // 创建一个http请求
        HttpURLConnection con = this.prepareConnection(extraHeaders);
        con.connect();
        OutputStream ops = con.getOutputStream();

        try {
            // 直接发起对应方法的调用
            super.invoke(methodName, argument, ops);
        } finally {
            ops.close();
        }

        try {
            InputStream ips = con.getInputStream();

            Object var8;
            try {
                var8 = super.readResponse(returnType, ips);
            } finally {
                ips.close();
            }

            return var8;
        } catch (IOException var17) {
            throw new HttpException(readString(con.getErrorStream()), var17);
        }
    }
}

这块有点绕，需要对JsonRpcServer、JettyHttpBinder有所了解，这块的知识点还是需要补一下的。

consumer请求之后，最终还是到了HttpProtocol中的InternalHandler来响应请求，如下所示

2.3 InternalHandler响应请求

private class InternalHandler implements HttpHandler {

    private boolean cors;

    public InternalHandler(boolean cors) {
        this.cors = cors;
    }

    @Override
    public void handle(HttpServletRequest request, HttpServletResponse response)
        throws ServletException {
        // 根据URI获取之前provider创建的JsonRpcServer
        String uri = request.getRequestURI();
        JsonRpcServer skeleton = skeletonMap.get(uri);
        if (cors) {
            response.setHeader(ACCESS_CONTROL_ALLOW_ORIGIN_HEADER, "*");
            response.setHeader(ACCESS_CONTROL_ALLOW_METHODS_HEADER, "POST");
            response.setHeader(ACCESS_CONTROL_ALLOW_HEADERS_HEADER, "*");
        }
        if (request.getMethod().equalsIgnoreCase("OPTIONS")) {
            response.setStatus(200);
        } else if (request.getMethod().equalsIgnoreCase("POST")) {
            RpcContext.getContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
            try {
                // 交由JsonRpcServer.handle()方法处理
                skeleton.handle(request.getInputStream(), response.getOutputStream());
            } catch (Throwable e) {
                throw new ServletException(e);
            }
        } else {
            response.setStatus(500);
        }
    }
}

注意：这里只允许post请求，get请求直接抛出500异常

请求交由JsonRpcServer处理这块逻辑，笔者不再深入，就是原生的JsonRpcServer相关处理即可。