gRPC 中泛化调用服务接口

gRPC 中泛化调用服务接口

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

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

大致的流程是:

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

实现

使用 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 one rror(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 对象;然后根据方法类型,使用MethodDescriptorCallOptions
发起请求;当接收到响应后将 DynamicMessage 解析为对应的格式的字符串;完成调用



参考文档

上一篇:gRPC 中打印请求二进制日志


下一篇:gRPC 中监听 Sream 和 Transport 的事件