gRPC 中泛化调用服务接口

January 29, 2021 · 454 words · 3 min

gRPC 中泛化调用服务接口

gRPC 没有直接支持泛化调用，protobuf 可以不依赖于生成的代码实现调用，所以可以通过反射接口间接实现泛化调用

要求 Server 端提供 grpc.reflection.v1alpha.ServerReflection 服务，用于获取服务的描述文件

大致的流程是：

根据方法名称，调用服务端反射服务的方法，获取方法所在 proto 文件的描述
根据 proto 描述文件，获取文件描述、服务描述，用于重新构建要被调用方法的方法描述 MethodDescriptor
根据方法描述，将请求内容序列化为对应的类型
使用重新构建的MethodDescriptor和其他参数对 Server 端相应的方法发起调用
解析响应并返回

实现

使用 JSON 格式请求被调用的服务方法，并返回 JSON 格式的响应

proto 定义

syntax = "proto3";

package io.github.helloworlde.grpc;

option go_package = "api;grpc_gateway";
option java_package = "io.github.helloworlde.grpc";
option java_multiple_files = true;
option java_outer_classname = "HelloWorldGrpc";

service HelloService{
  rpc SayHello(HelloMessage) returns (HelloResponse){
  }
}

message HelloMessage {
  string message = 2;
}

message HelloResponse {
  string message = 1;
}

调用

1. 构建反射服务 Stub

需要调用反射服务的方法，该方法是双向流

// 构建 Channel
ManagedChannel channel=ManagedChannelBuilder.forAddress("127.0.0.1",9090)
                                            .usePlaintext()
                                            .build();
// 使用 Channel 构建 BlockingStub
ServerReflectionGrpc.ServerReflectionStub reflectionStub=ServerReflectionGrpc.newStub(channel);
// 响应观察器
StreamObserver<ServerReflectionResponse> streamObserver=new StreamObserver<ServerReflectionResponse>(){
    @Override
    public void onNext(ServerReflectionResponse response){
        // 处理响应
    }

    @Override
    public void onError(Throwable t){

    }

    @Override
    public void onCompleted(){
        log.info("Complete");
    }
};
// 请求观察器
StreamObserver<ServerReflectionRequest> requestStreamObserver=reflectionStub.serverReflectionInfo(streamObserver);

2. 根据方法名称获取文件描述

这里的 methodSymbol 即服务或方法的限定名，可以是 package.service 或者 package.service.method ，如 io.github.helloworlde.grpc.HelloService.SayHello，需要注意方法前是 .不是/

// 构建并发送获取方法文件描述请求
ServerReflectionRequest getFileContainingSymbolRequest=ServerReflectionRequest.newBuilder()
        .setFileContainingSymbol(methodSymbol)
        .build();
        requestStreamObserver.onNext(getFileContainingSymbolRequest);

3. 处理响应，解析 FileDescriptor

返回的响应后会触发 onNext 方法，如果响应类型是文件描述类型，即 FILE_DESCRIPTOR_RESPONSE，则进行处理

public void onNext(ServerReflectionResponse response) {
    try {
        // 只需要关注文件描述类型的响应
        if (response.getMessageResponseCase() == ServerReflectionResponse.MessageResponseCase.FILE_DESCRIPTOR_RESPONSE) {
            List<ByteString> fileDescriptorProtoList = response.getFileDescriptorResponse().getFileDescriptorProtoList();
            handleResponse(fileDescriptorProtoList, channel, methodSymbol, requestContent);
        } else {
            log.warn("未知响应类型: " + response.getMessageResponseCase());
        }
    } catch (Exception e) {
        log.error("处理响应失败: {}", e.getMessage(), e);
    }
}

handleResponse

在处理请求时，先解析了包名、服务名和方法名，然后根据包名和服务名，从返回的文件描述中获取到了响应方法所在文件的描述；然后从文件描述中获取服务描述，最终获取到方法描述，根据方法描述执行调用

private static void handleResponse(List<ByteString> fileDescriptorProtoList,
                                   ManagedChannel channel,
                                   String methodFullName,
                                   String requestContent) {
    try {
        // 解析方法和服务名称
        String fullServiceName = extraPrefix(methodFullName);
        String methodName = extraSuffix(methodFullName);
        String packageName = extraPrefix(fullServiceName);
        String serviceName = extraSuffix(fullServiceName);

        // 根据响应解析 FileDescriptor
        Descriptors.FileDescriptor fileDescriptor = getFileDescriptor(fileDescriptorProtoList, packageName, serviceName);

        // 查找服务描述
        Descriptors.ServiceDescriptor serviceDescriptor = fileDescriptor.getFile().findServiceByName(serviceName);
        // 查找方法描述
        Descriptors.MethodDescriptor methodDescriptor = serviceDescriptor.findMethodByName(methodName);

        // 发起请求
        executeCall(channel, fileDescriptor, methodDescriptor, requestContent);
    } catch (Exception e) {
        log.error(e.getMessage(), e);
    }
}

getFileDescriptor

根据响应找到方法对应的文件的 FileDescriptorProto，然后构建出对应的 FileDescriptor

private static Descriptors.FileDescriptor getFileDescriptor(List<ByteString> fileDescriptorProtoList,
                                                            String packageName,
                                                            String serviceName) throws Exception {

    Map<String, DescriptorProtos.FileDescriptorProto> fileDescriptorProtoMap =
            fileDescriptorProtoList.stream()
                                   .map(bs -> {
                                       try {
                                           return DescriptorProtos.FileDescriptorProto.parseFrom(bs);
                                       } catch (InvalidProtocolBufferException e) {
                                           e.printStackTrace();
                                       }
                                       return null;
                                   })
                                   .filter(Objects::nonNull)
                                   .collect(Collectors.toMap(DescriptorProtos.FileDescriptorProto::getName, f -> f));


    if (fileDescriptorProtoMap.isEmpty()) {
        log.error("服务不存在");
        throw new IllegalArgumentException("方法的文件描述不存在");
    }

    // 查找服务对应的 Proto 描述
    DescriptorProtos.FileDescriptorProto fileDescriptorProto = findServiceFileDescriptorProto(packageName, serviceName, fileDescriptorProtoMap);

    // 获取这个 Proto 的依赖
    Descriptors.FileDescriptor[] dependencies = getDependencies(fileDescriptorProto, fileDescriptorProtoMap);

    // 生成 Proto 的 FileDescriptor
    return Descriptors.FileDescriptor.buildFrom(fileDescriptorProto, dependencies);
}

4. 执行调用

生成方法描述

在执行调用时，需要重新生成 MethodDescriptor；因为获取到的 MethodDescriptor 中的方法全名是package.service.method 格式，而需要的是package.service/method格式，同时请求和响应类型也需要重新设置为 DynamicMessage,所以需要重新生成 MethodDescriptor

private static MethodDescriptor<DynamicMessage, DynamicMessage> generateMethodDescriptor(Descriptors.MethodDescriptor originMethodDescriptor) {
    // 生成方法全名
    String fullMethodName = MethodDescriptor.generateFullMethodName(originMethodDescriptor.getService().getFullName(), originMethodDescriptor.getName());
    // 请求和响应类型
    MethodDescriptor.Marshaller<DynamicMessage> inputTypeMarshaller = ProtoUtils.marshaller(DynamicMessage.newBuilder(originMethodDescriptor.getInputType())
                                                                                                          .buildPartial());
    MethodDescriptor.Marshaller<DynamicMessage> outputTypeMarshaller = ProtoUtils.marshaller(DynamicMessage.newBuilder(originMethodDescriptor.getOutputType())
                                                                                                           .buildPartial());

    // 生成方法描述, originMethodDescriptor 的 fullMethodName 不正确
    return MethodDescriptor.<DynamicMessage, DynamicMessage>newBuilder()
            .setFullMethodName(fullMethodName)
            .setRequestMarshaller(inputTypeMarshaller)
            .setResponseMarshaller(outputTypeMarshaller)
            // 使用 UNKNOWN，自动修改
            .setType(MethodDescriptor.MethodType.UNKNOWN)
            .build();
}

执行调用

同时需要根据文件描述，将请求的类型转为对应的请求类型，生成 DynamicMessage 对象；然后根据方法类型，使用MethodDescriptor 和 CallOptions 发起请求；当接收到响应后将 DynamicMessage 解析为对应的格式的字符串；完成调用

参考文档

Thrift 中的 Protocol Thrfit 中的 Server