Rest 协议

本文将介绍 Dubbo 的 REST/HTTP 协议设计。

本文将介绍 Dubbo 的 REST/HTTP 协议设计。

RestProtocol 设计

原版本dubbo rest

consumer

restClient支持 依赖resteasy 不支持spring mvc

provider(较重)

依赖web container (tomcat,jetty，)servlet 模式，jaxrs netty server

新版本dubbo rest

更加轻量，具有dubbo风格的rest，微服务体系互通（Springcloud Alibaba）

1.注解解析

2.报文编解码

3.restClient

4.restServer(netty)

支持程度：

content-type text json xml form(后续会扩展)

注解

param,header,body,pathvaribale （spring mvc & resteasy）

Http 协议报文

POST /test/path?  HTTP/1.1
Host: localhost:8080
Connection: keep-alive
Content-type: application/json
{"name":"dubbo","age":10,"address":"hangzhou"}

dubbo http(header)

// service key header
path: com.demo.TestInterface
group: demo
port: 80
version: 1.0.0
// 保证长连接
Keep-Alive,Connection: keep-alive
Keep-alive: 60
// RPCContext Attachment
userId: 123456

支持粒度

rest注解解析

ServiceRestMetadataResolver

JAXRSServiceRestMetadataResolver
SpringMvcServiceRestMetadataResolver

ServiceRestMetadata

public class ServiceRestMetadata implements Serializable {
    private String serviceInterface; // com.demo.TestInterface
    private String version;// 1.0.0
    private String group;// demo
    private Set<RestMethodMetadata> meta;// method 元信息
    private int port;// 端口 for provider service key
    private boolean consumer;// consumer 标志
    /**
     * make a distinction between mvc & resteasy
     */
    private Class codeStyle;//
     /**
     *  for provider
     */
    private Map<PathMatcher, RestMethodMetadata> pathToServiceMap;
    /**
    * for consumer
    */
    private Map<String, Map<ParameterTypesComparator, RestMethodMetadata>> methodToServiceMa

RestMethodMetadata

public class RestMethodMetadata implements Serializable {
    private MethodDefinition method; // method 定义信息（name ,pramType,returnType）
    private RequestMetadata request;// 请求元信息
    private Integer urlIndex;
    private Integer bodyIndex;
    private Integer headerMapIndex;
    private String bodyType;
    private Map<Integer, Collection<String>> indexToName;
    private List<String> formParams;
    private Map<Integer, Boolean> indexToEncoded;
    private ServiceRestMetadata serviceRestMetadata;
    private List<ArgInfo> argInfos;
    private Method reflectMethod;
    /**
     *  make a distinction between mvc & resteasy
     */
    private Class codeStyle;

ArgInfo

public class ArgInfo {
    /**
     * method arg index 0,1,2,3
     */
    private int index;
    /**
     * method annotation name or name
     */
    private String annotationNameAttribute;
    /**
     * param annotation type
     */
    private Class paramAnnotationType;
    /**
     * param Type
     */
    private Class paramType;
    /**
     * param name
     */
    private String paramName;
    /**
     * url split("/") String[n]  index
     */
    private int urlSplitIndex;
    private Object defaultValue;
    private boolean formContentType;

RequestMeatadata

public class RequestMetadata implements Serializable {
    private static final long serialVersionUID = -240099840085329958L;
    private String method;// 请求method
    private String path;// 请求url
    private Map<String, List<String>> params // param参数?拼接
    private Map<String, List<String>> headers// header;
    private Set<String> consumes // content-type;
    private Set<String> produces // Accept;

Consumer 代码

refer

 @Override
    protected <T> Invoker<T> protocolBindingRefer(final Class<T> type, final URL url) throws RpcException {
        // restClient spi创建
        ReferenceCountedClient<? extends RestClient> refClient =
            clients.computeIfAbsent(url.getAddress(), key -> createReferenceCountedClient(url, clients));
        refClient.retain();
        // resolve metadata
        Map<String, Map<ParameterTypesComparator, RestMethodMetadata>> metadataMap = MetadataResolver.resolveConsumerServiceMetadata(type, url);
        ReferenceCountedClient<? extends RestClient> finalRefClient = refClient;
        Invoker<T> invoker = new AbstractInvoker<T>(type, url, new String[]{INTERFACE_KEY, GROUP_KEY, TOKEN_KEY}) {
            @Override
            protected Result doInvoke(Invocation invocation) {
                try {
                    // 获取 method的元信息
                    RestMethodMetadata restMethodMetadata = metadataMap.get(invocation.getMethodName()).get(ParameterTypesComparator.getInstance(invocation.getParameterTypes()));
                    RequestTemplate requestTemplate = new RequestTemplate(invocation, restMethodMetadata.getRequest().getMethod(), url.getAddress(), getContextPath(url));
                    HttpConnectionCreateContext httpConnectionCreateContext = new HttpConnectionCreateContext();
                    // TODO  dynamic load config
                    httpConnectionCreateContext.setConnectionConfig(new HttpConnectionConfig());
                    httpConnectionCreateContext.setRequestTemplate(requestTemplate);
                    httpConnectionCreateContext.setRestMethodMetadata(restMethodMetadata);
                    httpConnectionCreateContext.setInvocation(invocation);
                    httpConnectionCreateContext.setUrl(url);
                                        // http 信息构建拦截器
                    for (HttpConnectionPreBuildIntercept intercept : httpConnectionPreBuildIntercepts) {
                        intercept.intercept(httpConnectionCreateContext);
                    }
                    CompletableFuture<RestResult> future = finalRefClient.getClient().send(requestTemplate);
                    CompletableFuture<AppResponse> responseFuture = new CompletableFuture<>();
                    AsyncRpcResult asyncRpcResult = new AsyncRpcResult(responseFuture, invocation);
                    // response 处理
                    future.whenComplete((r, t) -> {
                        if (t != null) {
                            responseFuture.completeExceptionally(t);
                        } else {
                            AppResponse appResponse = new AppResponse();
                            try {
                                int responseCode = r.getResponseCode();
                                MediaType mediaType = MediaType.TEXT_PLAIN;
                                if (400 < responseCode && responseCode < 500) {
                                    throw new HttpClientException(r.getMessage());
                                } else if (responseCode >= 500) {
                                    throw new RemoteServerInternalException(r.getMessage());
                                } else if (responseCode < 400) {
                                    mediaType = MediaTypeUtil.convertMediaType(r.getContentType());
                                }
                                Object value = HttpMessageCodecManager.httpMessageDecode(r.getBody(),
                                    restMethodMetadata.getReflectMethod().getReturnType(), mediaType);
                                appResponse.setValue(value);
                                Map<String, String> headers = r.headers()
                                    .entrySet()
                                    .stream()
                                    .collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().get(0)));
                                appResponse.setAttachments(headers);
                                responseFuture.complete(appResponse);
                            } catch (Exception e) {
                                responseFuture.completeExceptionally(e);
                            }
                        }
                    });
                    return asyncRpcResult;
                } catch (RpcException e) {
                    if (e.getCode() == RpcException.UNKNOWN_EXCEPTION) {
                        e.setCode(getErrorCode(e.getCause()));
                    }
                    throw e;
                }
            }
            @Override
            public void destroy() {
                super.destroy();
                invokers.remove(this);
                destroyInternal(url);
            }
        };
        invokers.add(invoker);
        return invoker;

provider 代码

export

 public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
        URL url = invoker.getUrl();
        final String uri = serviceKey(url);
        Exporter<T> exporter = (Exporter<T>) exporterMap.get(uri);
        if (exporter != null) {
            // When modifying the configuration through override, you need to re-expose the newly modified service.
            if (Objects.equals(exporter.getInvoker().getUrl(), invoker.getUrl())) {
                return exporter;
            }
        }
        // TODO  addAll metadataMap to RPCInvocationBuilder metadataMap
        Map<PathMatcher, RestMethodMetadata> metadataMap = MetadataResolver.resolveProviderServiceMetadata(url.getServiceModel().getProxyObject().getClass(),url);
        PathAndInvokerMapper.addPathAndInvoker(metadataMap, invoker);
        final Runnable runnable = doExport(proxyFactory.getProxy(invoker, true), invoker.getInterface(), invoker.getUrl());
        exporter = new AbstractExporter<T>(invoker) {
            @Override
            public void afterUnExport() {
                exporterMap.remove(uri);
                if (runnable != null) {
                    try {
                        runnable.run();
                    } catch (Throwable t) {
                        logger.warn(PROTOCOL_UNSUPPORTED, "", "", t.getMessage(), t);
                    }
                }
            }
        };
        exporterMap.put(uri, exporter);
        return exporter;
    }

RestHandler

 private class RestHandler implements HttpHandler<HttpServletRequest, HttpServletResponse> {
        @Override
        public void handle(HttpServletRequest servletRequest, HttpServletResponse servletResponse) throws IOException, ServletException {
             // 有servlet reuqest 和nettyRequest
            RequestFacade request = RequestFacadeFactory.createRequestFacade(servletRequest);
            RpcContext.getServiceContext().setRemoteAddress(request.getRemoteAddr(), request.getRemotePort());
//            dispatcher.service(request, servletResponse);
            Pair<RpcInvocation, Invoker> build = null;
            try {
                // 根据请求信息创建 RPCInvocation
                build = RPCInvocationBuilder.build(request, servletRequest, servletResponse);
            } catch (PathNoFoundException e) {
                servletResponse.setStatus(404);
            }
            Invoker invoker = build.getSecond();
            Result invoke = invoker.invoke(build.getFirst());
            // TODO handling  exceptions
            if (invoke.hasException()) {
                servletResponse.setStatus(500);
            } else {
                try {
                    Object value = invoke.getValue();
                    String accept = request.getHeader(RestConstant.ACCEPT);
                    MediaType mediaType = MediaTypeUtil.convertMediaType(accept);
                    // TODO write response
                    HttpMessageCodecManager.httpMessageEncode(servletResponse.getOutputStream(), value, invoker.getUrl(), mediaType);
                    servletResponse.setStatus(200);
                } catch (Exception e) {
                    servletResponse.setStatus(500);
                }
            }
            // TODO add Attachment header
        }
    }

RPCInvocationBuilder

{
    private static final ParamParserManager paramParser = new ParamParserManager();
    public static Pair<RpcInvocation, Invoker> build(RequestFacade request, Object servletRequest, Object servletResponse) {
        // 获取invoker
        Pair<Invoker, RestMethodMetadata> invokerRestMethodMetadataPair = getRestMethodMetadata(request);
        RpcInvocation rpcInvocation = createBaseRpcInvocation(request, invokerRestMethodMetadataPair.getSecond());
        ProviderParseContext parseContext = createParseContext(request, servletRequest, servletResponse, invokerRestMethodMetadataPair.getSecond());
        // 参数构建
        Object[] args = paramParser.providerParamParse(parseContext);
        rpcInvocation.setArguments(args);
        return Pair.make(rpcInvocation, invokerRestMethodMetadataPair.getFirst());
    }
    private static ProviderParseContext createParseContext(RequestFacade request, Object servletRequest, Object servletResponse, RestMethodMetadata restMethodMetadata) {
        ProviderParseContext parseContext = new ProviderParseContext(request);
        parseContext.setResponse(servletResponse);
        parseContext.setRequest(servletRequest);
        Object[] objects = new Object[restMethodMetadata.getArgInfos().size()];
        parseContext.setArgs(Arrays.asList(objects));
        parseContext.setArgInfos(restMethodMetadata.getArgInfos());
        return parseContext;
    }
    private static RpcInvocation createBaseRpcInvocation(RequestFacade request, RestMethodMetadata restMethodMetadata) {
        RpcInvocation rpcInvocation = new RpcInvocation();
        int localPort = request.getLocalPort();
        String localAddr = request.getLocalAddr();
        int remotePort = request.getRemotePort();
        String remoteAddr = request.getRemoteAddr();
        String HOST = request.getHeader(RestConstant.HOST);
        String GROUP = request.getHeader(RestConstant.GROUP);
        String PATH = request.getHeader(RestConstant.PATH);
        String VERSION = request.getHeader(RestConstant.VERSION);
        String METHOD = restMethodMetadata.getMethod().getName();
        String[] PARAMETER_TYPES_DESC = restMethodMetadata.getMethod().getParameterTypes();
        rpcInvocation.setParameterTypes(restMethodMetadata.getReflectMethod().getParameterTypes());
        rpcInvocation.setMethodName(METHOD);
        rpcInvocation.setAttachment(RestConstant.GROUP, GROUP);
        rpcInvocation.setAttachment(RestConstant.METHOD, METHOD);
        rpcInvocation.setAttachment(RestConstant.PARAMETER_TYPES_DESC, PARAMETER_TYPES_DESC);
        rpcInvocation.setAttachment(RestConstant.PATH, PATH);
        rpcInvocation.setAttachment(RestConstant.VERSION, VERSION);
        rpcInvocation.setAttachment(RestConstant.HOST, HOST);
        rpcInvocation.setAttachment(RestConstant.REMOTE_ADDR, remoteAddr);
        rpcInvocation.setAttachment(RestConstant.LOCAL_ADDR, localAddr);
        rpcInvocation.setAttachment(RestConstant.REMOTE_PORT, remotePort);
        rpcInvocation.setAttachment(RestConstant.LOCAL_PORT, localPort);
        Enumeration<String> attachments = request.getHeaders(RestConstant.DUBBO_ATTACHMENT_HEADER);
        while (attachments != null && attachments.hasMoreElements()) {
            String s =  attachments.nextElement();
            String[] split = s.split("=");
            rpcInvocation.setAttachment(split[0], split[1]);
        }
        // TODO set path,version,group and so on
        return rpcInvocation;
    }
    private static Pair<Invoker, RestMethodMetadata> getRestMethodMetadata(RequestFacade request) {
        String path = request.getRequestURI();
        String version = request.getHeader(RestConstant.VERSION);
        String group = request.getHeader(RestConstant.GROUP);
        int port = request.getIntHeader(RestConstant.REST_PORT);
        return PathAndInvokerMapper.getRestMethodMetadata(path, version, group, port);
    }
}

编码示例

API

mvc

@RestController()
@RequestMapping("/demoService")
public interface DemoService {
    @RequestMapping(value = "/hello", method = RequestMethod.GET)
    Integer hello(@RequestParam Integer a, @RequestParam Integer b);
    @RequestMapping(value = "/error", method = RequestMethod.GET)
    String error();
    @RequestMapping(value = "/say", method = RequestMethod.POST, consumes = MediaType.TEXT_PLAIN_VALUE)
    String sayHello(@RequestBody String name);
}

resteasy:

@Path("/demoService")
public interface RestDemoService {
    @GET
    @Path("/hello")
    Integer hello(@QueryParam("a")Integer a,@QueryParam("b") Integer b);
    @GET
    @Path("/error")
    String error();
    @POST
    @Path("/say")
    @Consumes({MediaType.TEXT_PLAIN})
    String sayHello(String name);
    boolean isCalled();
}

impl(service)

@DubboService()
public class RestDemoServiceImpl implements RestDemoService {
    private static Map<String, Object> context;
    private boolean called;
    @Override
    public String sayHello(String name) {
        called = true;
        return "Hello, " + name;
    }
    public boolean isCalled() {
        return called;
    }
    @Override
    public Integer hello(Integer a, Integer b) {
        context = RpcContext.getServerAttachment().getObjectAttachments();
        return a + b;
    }
    @Override
    public String error() {
        throw new RuntimeException();
    }
    public static Map<String, Object> getAttachments() {
        return context;
    }
}

流程图

Consumer

Provider(RestServer)

场景

1.体系互通

非dubbo体系互通（Springcloud alibaba 互通）

互通条件：

2.dubbo 双注册

完成应用级别注册，（dubo2-dubbo3 过度），dubbo版本升级

3.多协议发布

配置：

<dubbo:service interface="org.apache.dubbo.samples.DemoService" protocol="dubbo, grpc,rest"/>

4.跨语言

5.多协议交互

6.协议迁移

rest编码风格

Http协议更通用跨语言调用

dubbo rest 对其他http服务 进行调用

其他httpclient 对dubbo rest进行调用

dubbo restServer 可以与其他web服务，浏览器等客户端直接进行http交互

consumer TODOLIST

功能已经初步实现，可以调通解析response

1. org/apache/dubbo/rpc/protocol/rest/RestProtocol.java:157 dynamic load config

2.org/apache/dubbo/remoting/http/factory/AbstractHttpClientFactory.java:50 load config HttpClientConfig

3.org/apache/dubbo/rpc/protocol/rest/annotation/metadata/MetadataResolver.java:52 support Dubbo style service

4.org/apache/dubbo/remoting/http/restclient/HttpClientRestClient.java:120 TODO config

5.org/apache/dubbo/remoting/http/restclient/HttpClientRestClient.java:140 TODO close judge

6.org/apache/dubbo/rpc/protocol/rest/message/decode/MultiValueCodec.java:35 TODO java bean get set convert

provider TODOLIST

待实现

基于netty实现支持http协议的NettyServer

无注解协议定义

官网场景补充

Rest使用说明文档及demo

最后修改 April 15, 2023: Update protocol-http.md (#2535) (c2f579b2413)