• grpc Java demo与Springboot改造支持grpc通信

    前言

    最近调研grpc的情况，发现grpc实际上还是HTTP2协议，实际上就是http2+proto传输。那么是否可以在现有的server支持呢，试了下，还真可以，但是笔者在返回数据时有个问题一直没有思路。

    grpc原生demo

    原生的grpc-java很简单，实际上开源的google原生包和grpc-Spring-boot-starter都有成熟的开源方案，以net.devh为例

    grpc-api

    1.     <properties>
    2.         <protobuf.version>3.19.1protobuf.version>
    3.         <protobuf-plugin.version>0.6.1protobuf-plugin.version>
    4.         <grpc.version>1.48.1grpc.version>
    5.     properties>
    6.  
    7.     <dependencies>
    8.         <dependency>
    9.             <groupId>io.grpcgroupId>
    10.             <artifactId>grpc-stubartifactId>
    11.             <version>${grpc.version}version>
    12.         dependency>
    13.         <dependency>
    14.             <groupId>io.grpcgroupId>
    15.             <artifactId>grpc-protobufartifactId>
    16.             <version>${grpc.version}version>
    17.         dependency>
    18.         <dependency>
    19.             
    20.             <groupId>jakarta.annotationgroupId>
    21.             <artifactId>jakarta.annotation-apiartifactId>
    22.             <version>1.3.5version>
    23.             <optional>trueoptional>
    24.         dependency>
    25.     dependencies>
    26.  
    27.     <build>
    28.         <extensions>
    29.             <extension>
    30.                 <groupId>kr.motd.mavengroupId>
    31.                 <artifactId>os-maven-pluginartifactId>
    32.                 <version>1.7.0version>
    33.             extension>
    34.         extensions>
    35.  
    36.         <plugins>
    37.             <plugin>
    38.                 <groupId>org.xolstice.maven.pluginsgroupId>
    39.                 <artifactId>protobuf-maven-pluginartifactId>
    40.                 <version>${protobuf-plugin.version}version>
    41.                 <configuration>
    42.                     <protocArtifact>com.google.protobuf:protoc:${protobuf.version}:exe:${os.detected.classifier}protocArtifact>
    43.                     <pluginId>grpc-javapluginId>
    44.                     <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}pluginArtifact>
    45.                 configuration>
    46.                 <executions>
    47.                     <execution>
    48.                         <goals>
    49.                             <goal>compilegoal>
    50.                             <goal>compile-customgoal>
    51.                         goals>
    52.                     execution>
    53.                 executions>
    54.             plugin>
    55.         plugins>
    56.     build>

    然后编写proto文件，在src/main/proto下编写

    1. syntax = "proto3";
    2.  
    3. package com.feng.proto.api;
    4.  
    5. option java_multiple_files = true;
    6. option java_package = "com.feng.proto.api.lib";
    7. option java_outer_classname = "HelloWorld";
    8.  
    9. // The greeting service definition.
    10. service HiService {
    11.   // Sends a greeting
    12.   rpc SayHello (HelloRequest) returns (HelloReply) {
    13.   }
    14.  
    15.   rpc sayAge(HttpDemoRequest) returns (HttpDemoReply) {
    16.  
    17.   }
    18. }
    19.  
    20. // The request message containing the user's name.
    21. message HelloRequest {
    22.   string name = 1;
    23. }
    25. message HttpDemoRequest {
    26.   string name = 1;
    27.   int32 age = 2;
    28. }
    30. // The response message containing the greetings
    31. message HelloReply {
    32.   string message = 1;
    33. }
    35. message HttpDemoReply {
    36.   string message = 1;
    37.   int32 age = 2;
    38. }

    关键点，包名+服务名+方法名是调用的关键，参数与返回值是grpc特定的proto协议，普通的proto是无法处理的

     

     编译成java源码和class文件

     

    grpc-server

    依赖刚刚创建的grpc-api的module

    1.         <dependency>
    2.             <groupId>net.devhgroupId>
    3.             <artifactId>grpc-server-spring-boot-starterartifactId>
    4.             <version>2.13.1.RELEASEversion>
    5.         dependency>
    6.         <dependency>
    7.             <groupId>org.examplegroupId>
    8.             <artifactId>grpc-interfaceartifactId>
    9.             <version>1.0-SNAPSHOTversion>
    10.         dependency>
    11.

    编写grpc实现逻辑

    1. @GrpcService
    2. public class MyServiceImpl extends HiServiceGrpc.HiServiceImplBase {
    3.     @Override
    4.     public void sayHello(HelloRequest req, StreamObserver responseObserver) {
    5.         HelloReply reply = HelloReply.newBuilder().setMessage("Hello ==> " + req.getName()).build();
    6.         responseObserver.onNext(reply);
    7.         responseObserver.onCompleted();
    8.     }
    9.  
    10. }
    11.  
    12. @SpringBootApplication
    13. public class ServerMain {
    14.     public static void main(String[] args) {
    15.         SpringApplication.run(ServerMain.class, args);
    16.     }
    17. }

     

    grpc-client

    boot web自行依赖，关键依赖如下：

    1.         <dependency>
    2.             <groupId>org.examplegroupId>
    3.             <artifactId>grpc-interfaceartifactId>
    4.             <version>1.0-SNAPSHOTversion>
    5.         dependency>
    6.         <dependency>
    7.             <groupId>net.devhgroupId>
    8.             <artifactId>grpc-client-spring-boot-starterartifactId>
    9.             <version>2.13.1.RELEASEversion>
    10.         dependency>

    实现调用

    1. @Service
    2. public class MyServiceClient {
    3.  
    4.     @GrpcClient("myClient")
    5.     private HiServiceGrpc.HiServiceBlockingStub stub;
    6.  
    7.     public String sayHello(String name) {
    8.         HelloRequest request = HelloRequest.newBuilder()
    9.                 .setName(name)
    10.                 .build();
    11.         return stub.sayHello(request).getMessage();
    12.     }
    13.  
    14.     public int sayAge(int age){
    15.         HttpDemoRequest request = HttpDemoRequest.newBuilder()
    16.                 .setName("tom")
    17.                 .setAge(age)
    18.                 .build();
    19.         return stub.sayAge(request).getAge();
    20.     }
    21.  
    22. }
    23.  
    24. @SpringBootApplication
    25. @RestController
    26. public class ClientMain {
    27.     public static void main(String[] args) {
    28.         SpringApplication.run(ClientMain.class, args);
    29.     }
    30.  
    31.     @Autowired
    32.     private MyServiceClient myServiceClient;
    33.  
    34.     @RequestMapping("/hello")
    35.     public String call(){
    36.  
    37.         String res = myServiceClient.sayHello("str222");
    38.         return res;
    39.     }
    40.  
    41.     @RequestMapping("/hello2")
    42.     public Integer callAge(){
    43.         int res = myServiceClient.sayAge(11);
    44.         return res;
    45.     }
    46. }

    配置端口（本机需要），server的url，毕竟没接入注册中心，而且H2的SSL/TLS实际上，内部环境用途不大，适合对外使用，可以关闭，使用H2C。

    1. server.port = 8082
    2. grpc.client.myClient.address=static://localhost:9090
    3. grpc.client.myClient.negotiationType=PLAINTEXT

    访问localhost:8082/hello 

     wireshark抓包

    wireshark默认不识别http2，需要配置，以macOS为例，使用sudo chown -R xxx:admin /dev/bpf*

    打开wireshark，Analyze Decode As

    编辑协议，端口，

     

    然后对lo环回网卡抓包

     

    可以读取到请求和返回包，可以看到url header body

    可以看到url就是包名.服务名/方法名，header是特定义的，body实际上也是特殊的proto协议 

     再看看返回结果

    返回居然在Response后再次发送了Header，这个是笔者在使用普通Tomcat处理时尚未解决的地方

     

     

    springboot支持grpc协议

    实际上Springboot原生支持protobuf协议，跟http 1.1还是2.0无关，仿造写一个支持grpc的，因为grpc的协议是定制的，所以需要一些包

    2.         
    3.             org.springframework.boot
    4.             spring-boot-starter-web
    5.             2.7.5
    6.         
    7.         
    8.             org.example
    9.             grpc-interface
    10.             1.0-SNAPSHOT
    11.         
    12.         
    13.             com.google.protobuf
    14.             protobuf-java-util
    15.             3.21.7
    16.         
    17.         
    18.             com.googlecode.protobuf-java-format
    19.             protobuf-java-format
    20.             1.4
    21.         
    22.  
    23.

    converter代码 

    1. package org.springframework.http.converter.protobuf;
    2.  
    3. import com.google.protobuf.*;
    4. import com.google.protobuf.util.JsonFormat;
    5. import com.googlecode.protobuf.format.FormatFactory;
    6. import com.googlecode.protobuf.format.ProtobufFormatter;
    7. import io.grpc.protobuf.lite.ProtoLiteUtils;
    8. import org.springframework.http.HttpHeaders;
    9. import org.springframework.http.HttpInputMessage;
    10. import org.springframework.http.HttpOutputMessage;
    11. import org.springframework.http.MediaType;
    12. import org.springframework.http.converter.AbstractHttpMessageConverter;
    13. import org.springframework.http.converter.HttpMessageConversionException;
    14. import org.springframework.http.converter.HttpMessageNotReadableException;
    15. import org.springframework.http.converter.HttpMessageNotWritableException;
    16. import org.springframework.lang.Nullable;
    17. import org.springframework.util.Assert;
    18. import org.springframework.util.ClassUtils;
    19. import org.springframework.util.ConcurrentReferenceHashMap;
    20.  
    21. import java.io.*;
    22. import java.lang.reflect.InvocationTargetException;
    23. import java.lang.reflect.Method;
    24. import java.nio.charset.Charset;
    25. import java.nio.charset.StandardCharsets;
    26. import java.util.Arrays;
    27. import java.util.Map;
    28.  
    29. import static org.springframework.http.MediaType.*;
    30. import static org.springframework.http.MediaType.APPLICATION_JSON;
    31.  
    32. public class ProtobufGrpcFormatHttpMessageConverter extends AbstractHttpMessageConverter {
    33.  
    34.     /**
    35.      * The default charset used by the converter.
    36.      */
    37.     public static final Charset DEFAULT_CHARSET = StandardCharsets.UTF_8;
    38.  
    39.     /**
    40.      * The media-type for protobuf {@code application/x-protobuf}.
    41.      */
    42.     public static final MediaType PROTOBUF = new MediaType("application", "grpc", DEFAULT_CHARSET);
    43.  
    44.     /**
    45.      * The HTTP header containing the protobuf schema.
    46.      */
    47.     public static final String X_PROTOBUF_SCHEMA_HEADER = "X-Protobuf-Schema";
    48.  
    49.     /**
    50.      * The HTTP header containing the protobuf message.
    51.      */
    52.     public static final String X_PROTOBUF_MESSAGE_HEADER = "X-Protobuf-Message";
    53.  
    54.  
    55.     private static final Map, Method> methodCache = new ConcurrentReferenceHashMap<>();
    56.  
    57.     final ExtensionRegistry extensionRegistry;
    58.  
    59.     @Nullable
    60.     private final ProtobufFormatSupport protobufFormatSupport;
    61.  
    62.  
    63.     /**
    64.      * Construct a new {@code ProtobufGrpcFormatHttpMessageConverter}.
    65.      */
    66.     public ProtobufGrpcFormatHttpMessageConverter() {
    67.         this(null, null);
    68.     }
    69.  
    70.     /**
    71.      * Construct a new {@code ProtobufGrpcFormatHttpMessageConverter} with an
    72.      * initializer that allows the registration of message extensions.
    73.      * @param registryInitializer an initializer for message extensions
    74.      * @deprecated as of Spring Framework 5.1, use {@link #ProtobufGrpcFormatHttpMessageConverter(ExtensionRegistry)} instead
    75.      */
    76.     @Deprecated
    77.     public ProtobufGrpcFormatHttpMessageConverter(@Nullable ExtensionRegistryInitializer registryInitializer) {
    78.         this(null, null);
    79.         if (registryInitializer != null) {
    80.             registryInitializer.initializeExtensionRegistry(this.extensionRegistry);
    81.         }
    82.     }
    83.  
    84.     /**
    85.      * Construct a new {@code ProtobufGrpcFormatHttpMessageConverter} with a registry that specifies
    86.      * protocol message extensions.
    87.      * @param extensionRegistry the registry to populate
    88.      */
    89.     public ProtobufGrpcFormatHttpMessageConverter(ExtensionRegistry extensionRegistry) {
    90.         this(null, extensionRegistry);
    91.     }
    92.  
    93.     ProtobufGrpcFormatHttpMessageConverter(@Nullable ProtobufFormatSupport formatSupport,
    94.                                  @Nullable ExtensionRegistry extensionRegistry) {
    95.  
    96.         if (formatSupport != null) {
    97.             this.protobufFormatSupport = formatSupport;
    98.         }
    99.         else if (ClassUtils.isPresent("com.googlecode.protobuf.format.FormatFactory", getClass().getClassLoader())) {
    100.             this.protobufFormatSupport = new ProtobufJavaFormatSupport();
    101.         }
    102.         else if (ClassUtils.isPresent("com.google.protobuf.util.JsonFormat", getClass().getClassLoader())) {
    103.             this.protobufFormatSupport = new ProtobufJavaUtilSupport(null, null);
    104.         }
    105.         else {
    106.             this.protobufFormatSupport = null;
    107.         }
    108.  
    109.         setSupportedMediaTypes(Arrays.asList(this.protobufFormatSupport != null ?
    110.                 this.protobufFormatSupport.supportedMediaTypes() : new MediaType[] {PROTOBUF, TEXT_PLAIN}));
    111.  
    112.         this.extensionRegistry = (extensionRegistry == null ? ExtensionRegistry.newInstance() : extensionRegistry);
    113.     }
    114.  
    115.  
    116.     @Override
    117.     protected boolean supports(Class clazz) {
    118.         return Message.class.isAssignableFrom(clazz);
    119.     }
    120.  
    121.     @Override
    122.     protected MediaType getDefaultContentType(Message message) {
    123.         return PROTOBUF;
    124.     }
    125.  
    126.     @Override
    127.     protected Message readInternal(Class clazz, HttpInputMessage inputMessage)
    128.             throws IOException, HttpMessageNotReadableException {
    129.  
    130.         MediaType contentType = inputMessage.getHeaders().getContentType();
    131.         if (contentType == null) {
    132.             contentType = PROTOBUF;
    133.         }
    134.         Charset charset = contentType.getCharset();
    135.         if (charset == null) {
    136.             charset = DEFAULT_CHARSET;
    137.         }
    138.  
    139.         Message.Builder builder = getMessageBuilder(clazz);
    140.         if (PROTOBUF.isCompatibleWith(contentType)) {
    141.             try {
    142.                 Method method = clazz.getDeclaredMethod("getDefaultInstance");
    143.                 MessageLite defaultInstance = (MessageLite) method.invoke(null);
    144.                 MessageLite messageLite = ProtoLiteUtils.marshaller(defaultInstance).parse(new BufferInputStream(inputMessage.getBody()));
    145.                 return (Message) messageLite;
    146.             } catch (NoSuchMethodException e) {
    147.                 throw new RuntimeException(e);
    148.             } catch (InvocationTargetException e) {
    149.                 throw new RuntimeException(e);
    150.             } catch (IllegalAccessException e) {
    151.                 throw new RuntimeException(e);
    152.             }
    153.  
    154.             //builder.mergeFrom(inputMessage.getBody(), this.extensionRegistry);
    155.         }
    156.         else if (TEXT_PLAIN.isCompatibleWith(contentType)) {
    157.             InputStreamReader reader = new InputStreamReader(inputMessage.getBody(), charset);
    158.             TextFormat.merge(reader, this.extensionRegistry, builder);
    159.         }
    160.         else if (this.protobufFormatSupport != null) {
    161.             this.protobufFormatSupport.merge(
    162.                     inputMessage.getBody(), charset, contentType, this.extensionRegistry, builder);
    163.         }
    164.         return builder.build();
    165.     }
    166.  
    167.     /**
    168.      * Create a new {@code Message.Builder} instance for the given class.
    169.      * 
      This method uses a ConcurrentReferenceHashMap for caching method lookups.
    170.      */
    171.     private Message.Builder getMessageBuilder(Class clazz) {
    172.         try {
    173.             Method method = methodCache.get(clazz);
    174.             if (method == null) {
    175.                 method = clazz.getMethod("newBuilder");
    176.                 methodCache.put(clazz, method);
    177.             }
    178.             return (Message.Builder) method.invoke(clazz);
    179.         }
    180.         catch (Exception ex) {
    181.             throw new HttpMessageConversionException(
    182.                     "Invalid Protobuf Message type: no invocable newBuilder() method on " + clazz, ex);
    183.         }
    184.     }
    185.  
    186.  
    187.     @Override
    188.     protected boolean canWrite(@Nullable MediaType mediaType) {
    189.         return (super.canWrite(mediaType) ||
    190.                 (this.protobufFormatSupport != null && this.protobufFormatSupport.supportsWriteOnly(mediaType)));
    191.     }
    192.  
    193.     @SuppressWarnings("deprecation")
    194.     @Override
    195.     protected void writeInternal(Message message, HttpOutputMessage outputMessage)
    196.             throws IOException, HttpMessageNotWritableException {
    197.  
    198.         MediaType contentType = outputMessage.getHeaders().getContentType();
    199.         if (contentType == null) {
    200.             contentType = getDefaultContentType(message);
    201.             Assert.state(contentType != null, "No content type");
    202.         }
    203.         Charset charset = contentType.getCharset();
    204.         if (charset == null) {
    205.             charset = DEFAULT_CHARSET;
    206.         }
    207.  
    208.         if (PROTOBUF.isCompatibleWith(contentType)) {
    209.             outputMessage.getHeaders().add("grpc-encoding", "identity");
    210.             outputMessage.getHeaders().add("grpc-accept-encoding", "gzip");
    211. //            outputMessage.getHeaders().add("grpc-status", "0");
    212.             //setProtoHeader(outputMessage, message);
    213.             CodedOutputStream codedOutputStream = CodedOutputStream.newInstance(outputMessage.getBody());
    214.             MessageLite messageLite = message;
    215.             int length = messageLite.getSerializedSize();
    216.             byte[] pre = LengthBytesUtils.intToBytes(length, 5);
    217.             MessageOutputStreamUtils.writePrefix(codedOutputStream, pre);
    218.             message.writeTo(codedOutputStream);
    219.  
    220. //            HttpHeaders headers = new HttpHeaders();
    221. //            headers.add("grpc-status", "0");
    222.             outputMessage.getHeaders().add("grpc-status", "0");
    223. //            headers.get
    224.             codedOutputStream.flush();
    225. //            MessageOutputStreamUtils.writeSuffix(codedOutputStream);
    226. //            codedOutputStream.flush();
    227.         }
    228.         else if (TEXT_PLAIN.isCompatibleWith(contentType)) {
    229.             OutputStreamWriter outputStreamWriter = new OutputStreamWriter(outputMessage.getBody(), charset);
    230.             TextFormat.print(message, outputStreamWriter);  // deprecated on Protobuf 3.9
    231.             outputStreamWriter.flush();
    232.             outputMessage.getBody().flush();
    233.         }
    234.         else if (this.protobufFormatSupport != null) {
    235.             this.protobufFormatSupport.print(message, outputMessage.getBody(), contentType, charset);
    236.             outputMessage.getBody().flush();
    237.         }
    238.     }
    239.  
    240.     /**
    241.      * Set the "X-Protobuf-*" HTTP headers when responding with a message of
    242.      * content type "application/x-protobuf"
    243.      * 
      Note: outputMessage.getBody() should not have been called
    244.      * before because it writes HTTP headers (making them read only).
    245.      */
    246.     private void setProtoHeader(HttpOutputMessage response, Message message) {
    247.         response.getHeaders().set(X_PROTOBUF_SCHEMA_HEADER, message.getDescriptorForType().getFile().getName());
    248.         response.getHeaders().set(X_PROTOBUF_MESSAGE_HEADER, message.getDescriptorForType().getFullName());
    249.     }
    250.  
    251.  
    252.     /**
    253.      * Protobuf format support.
    254.      */
    255.     interface ProtobufFormatSupport {
    256.  
    257.         MediaType[] supportedMediaTypes();
    258.  
    259.         boolean supportsWriteOnly(@Nullable MediaType mediaType);
    260.  
    261.         void merge(InputStream input, Charset charset, MediaType contentType,
    262.                    ExtensionRegistry extensionRegistry, Message.Builder builder)
    263.                 throws IOException, HttpMessageConversionException;
    264.  
    265.         void print(Message message, OutputStream output, MediaType contentType, Charset charset)
    266.                 throws IOException, HttpMessageConversionException;
    267.     }
    268.  
    269.  
    270.     /**
    271.      * {@link ProtobufFormatSupport} implementation used when
    272.      * {@code com.googlecode.protobuf.format.FormatFactory} is available.
    273.      */
    274.     static class ProtobufJavaFormatSupport implements ProtobufFormatSupport {
    275.  
    276.         private final ProtobufFormatter jsonFormatter;
    277.  
    278.         private final ProtobufFormatter xmlFormatter;
    279.  
    280.         private final ProtobufFormatter htmlFormatter;
    281.  
    282.         public ProtobufJavaFormatSupport() {
    283.             FormatFactory formatFactory = new FormatFactory();
    284.             this.jsonFormatter = formatFactory.createFormatter(FormatFactory.Formatter.JSON);
    285.             this.xmlFormatter = formatFactory.createFormatter(FormatFactory.Formatter.XML);
    286.             this.htmlFormatter = formatFactory.createFormatter(FormatFactory.Formatter.HTML);
    287.         }
    288.  
    289.         @Override
    290.         public MediaType[] supportedMediaTypes() {
    291.             return new MediaType[] {PROTOBUF, TEXT_PLAIN, APPLICATION_XML, APPLICATION_JSON};
    292.         }
    293.  
    294.         @Override
    295.         public boolean supportsWriteOnly(@Nullable MediaType mediaType) {
    296.             return TEXT_HTML.isCompatibleWith(mediaType);
    297.         }
    298.  
    299.         @Override
    300.         public void merge(InputStream input, Charset charset, MediaType contentType,
    301.                           ExtensionRegistry extensionRegistry, Message.Builder builder)
    302.                 throws IOException, HttpMessageConversionException {
    303.  
    304.             if (contentType.isCompatibleWith(APPLICATION_JSON)) {
    305.                 this.jsonFormatter.merge(input, charset, extensionRegistry, builder);
    306.             }
    307.             else if (contentType.isCompatibleWith(APPLICATION_XML)) {
    308.                 this.xmlFormatter.merge(input, charset, extensionRegistry, builder);
    309.             }
    310.             else {
    311.                 throw new HttpMessageConversionException(
    312.                         "protobuf-java-format does not support parsing " + contentType);
    313.             }
    314.         }
    315.  
    316.         @Override
    317.         public void print(Message message, OutputStream output, MediaType contentType, Charset charset)
    318.                 throws IOException, HttpMessageConversionException {
    319.  
    320.             if (contentType.isCompatibleWith(APPLICATION_JSON)) {
    321.                 this.jsonFormatter.print(message, output, charset);
    322.             }
    323.             else if (contentType.isCompatibleWith(APPLICATION_XML)) {
    324.                 this.xmlFormatter.print(message, output, charset);
    325.             }
    326.             else if (contentType.isCompatibleWith(TEXT_HTML)) {
    327.                 this.htmlFormatter.print(message, output, charset);
    328.             }
    329.             else {
    330.                 throw new HttpMessageConversionException(
    331.                         "protobuf-java-format does not support printing " + contentType);
    332.             }
    333.         }
    334.     }
    335.  
    336.  
    337.     /**
    338.      * {@link ProtobufFormatSupport} implementation used when
    339.      * {@code com.google.protobuf.util.JsonFormat} is available.
    340.      */
    341.     static class ProtobufJavaUtilSupport implements ProtobufFormatSupport {
    342.  
    343.         private final JsonFormat.Parser parser;
    344.  
    345.         private final JsonFormat.Printer printer;
    346.  
    347.         public ProtobufJavaUtilSupport(@Nullable JsonFormat.Parser parser, @Nullable JsonFormat.Printer printer) {
    348.             this.parser = (parser != null ? parser : JsonFormat.parser());
    349.             this.printer = (printer != null ? printer : JsonFormat.printer());
    350.         }
    351.  
    352.         @Override
    353.         public MediaType[] supportedMediaTypes() {
    354.             return new MediaType[] {PROTOBUF, TEXT_PLAIN, APPLICATION_JSON};
    355.         }
    356.  
    357.         @Override
    358.         public boolean supportsWriteOnly(@Nullable MediaType mediaType) {
    359.             return false;
    360.         }
    361.  
    362.         @Override
    363.         public void merge(InputStream input, Charset charset, MediaType contentType,
    364.                           ExtensionRegistry extensionRegistry, Message.Builder builder)
    365.                 throws IOException, HttpMessageConversionException {
    366.  
    367.             if (contentType.isCompatibleWith(APPLICATION_JSON)) {
    368.                 InputStreamReader reader = new InputStreamReader(input, charset);
    369.                 this.parser.merge(reader, builder);
    370.             }
    371.             else {
    372.                 throw new HttpMessageConversionException(
    373.                         "protobuf-java-util does not support parsing " + contentType);
    374.             }
    375.         }
    376.  
    377.         @Override
    378.         public void print(Message message, OutputStream output, MediaType contentType, Charset charset)
    379.                 throws IOException, HttpMessageConversionException {
    380.  
    381.             if (contentType.isCompatibleWith(APPLICATION_JSON)) {
    382.                 OutputStreamWriter writer = new OutputStreamWriter(output, charset);
    383.                 this.printer.appendTo(message, writer);
    384.                 writer.flush();
    385.             }
    386.             else {
    387.                 throw new HttpMessageConversionException(
    388.                         "protobuf-java-util does not support printing " + contentType);
    389.             }
    390.         }
    391.     }
    392.  
    393. }

     核心关键点

    并改造write和read，上面的代码已经改造了

    1. @RestController
    2. public class GrpcController {
    3.  
    4.     @RequestMapping(value = "/com.feng.proto.api.HiService/SayHello", method = RequestMethod.POST, produces = "application/grpc")
    5.     public HelloReply sayHello(@RequestBody HelloRequest helloRequest) {
    6.         HelloReply helloReply = HelloReply.newBuilder()
    7.                 .setMessage("hello ---- " + helloRequest.getName())
    8.                 .build();
    9.         return helloReply;
    10.     }
    11.  
    12.     @RequestMapping(value = "/com.feng.proto.api.HiService/sayAge", method = RequestMethod.POST, produces = "application/grpc")
    13.     public HttpDemoReply sayAge(@RequestBody HttpDemoRequest helloRequest) {
    14.         HttpDemoReply helloReply = HttpDemoReply.newBuilder()
    15.                 .setMessage("hello ---- " + helloRequest.getName())
    16.                 .setAge(helloRequest.getAge())
    17.                 .build();
    18.         return helloReply;
    19.     }
    20. }
    21.  
    22. @Configuration
    23. public class ConfigGrpcConverter {
    24.     @Bean
    25.     public ProtobufGrpcFormatHttpMessageConverter initProtobufGrpcFormatHttpMessageConverter(){
    26.         return new ProtobufGrpcFormatHttpMessageConverter();
    27.     }
    28. }

    抓包可以发现返回跟原生的grpc返回少了最后的http2发送header的逻辑

     请求包是一样的

     返回就不对了，缺少了grpc-status 0的header通过http2写回来。

     导致了，grpc的client端，实际上Response的body已经解析成功了，只不过结束符判断不对

     

    client端已经读取到消息了，并正常解析 

     

    grpc的protocol格式

    grpc的传输二进制不是传统的protobuf，在这个基础上定制了

    是：5个字节数组+protobuf字节数组

    这5个字节数组是protobuf的length int转为byte[]

    length长度算法

    而且谷歌的长度转换逻辑是定制的，从0~126再到-127~-1，共254个，那么进制就是253.

    1. package org.springframework.http.converter.protobuf;
    2.  
    3. public class LengthBytesUtils {
    4.     public static byte[] intToBytes(int length, int size) {
    5.         byte[] bytes = new byte[size];
    6.         int temp;
    7.         for (int i = size - 1; i > -1; i--) {
    8.             temp = (int) ((length / (Math.pow(253, (size - 1 - i)))) % 253);
    9.             if (temp > 127) {
    10.                 temp = -128 + 1 + temp - 127 + 1;
    11.             }
    12.             bytes[i] = (byte) temp;
    13.         }
    14.         return bytes;
    15.     }
    16.  
    17.     public static int bytesToInt(byte[] bytes) {
    18.         int length = 0;
    19.         int size = bytes.length;
    20.         for (int i = 0; i < size; i++) {
    21.             if (bytes[i] == 0){
    22.                 continue;
    23.             }
    24.             if (bytes[i] > 0) {
    25.                 length += bytes[i] * Math.pow(253, size-1-i);
    26.             } else {
    27.                 length += (127-1 + bytes[i] + 128-1) * Math.pow(253, size-1-i);
    28.             }
    29.         }
    30.         return length;
    31.     }
    32. }

    为此特意处理输入输出流

    1. package org.springframework.http.converter.protobuf;
    2.  
    3. import com.google.common.base.Preconditions;
    4. import io.grpc.Detachable;
    5. import io.grpc.HasByteBuffer;
    6. import io.grpc.KnownLength;
    7.  
    8. import javax.annotation.Nullable;
    9. import java.io.IOException;
    10. import java.io.InputStream;
    11. import java.nio.ByteBuffer;
    12.  
    13. public class BufferInputStream extends InputStream
    14.         implements KnownLength, HasByteBuffer, Detachable {
    15.     private InputStream buffer;
    16.  
    17.     public BufferInputStream(InputStream buffer) {
    18.         this.buffer = Preconditions.checkNotNull(buffer, "buffer");
    19.     }
    20.  
    21.     @Override
    22.     public int available() throws IOException {
    23.         byte[] lengthBytes = new byte[5];
    24.         buffer.read(lengthBytes, 0, 5);
    25.         int length = LengthBytesUtils.bytesToInt(lengthBytes);
    26.         byte[] valid = LengthBytesUtils.intToBytes(length, 5);
    27. //        buffer.skip(2);
    28.         return length;
    29.     }
    30.  
    31.     @Override
    32.     public int read() throws IOException {
    33.         if (buffer.read() == 0) {
    34.             // EOF.
    35.             return -1;
    36.         }
    37.         return buffer.read();
    38.     }
    39.  
    40.     @Override
    41.     public int read(byte[] dest, int destOffset, int length) throws IOException {
    42.         buffer.read(dest, destOffset, length);
    43.         return length;
    44.     }
    45.  
    46.     @Override
    47.     public long skip(long n) throws IOException {
    48.         buffer.skip(n);
    49.         return n;
    50.     }
    51.  
    52.     @Override
    53.     public void mark(int readlimit) {
    54.         buffer.mark(readlimit);
    55.     }
    56.  
    57.     @Override
    58.     public void reset() throws IOException {
    59.         buffer.reset();
    60.     }
    61.  
    62.     @Override
    63.     public boolean markSupported() {
    64.         return buffer.markSupported();
    65.     }
    66.  
    67.     @Override
    68.     public boolean byteBufferSupported() {
    69.         return false;
    70.     }
    71.  
    72.     @Nullable
    73.     @Override
    74.     public ByteBuffer getByteBuffer() {
    75.         return null;
    76.     }
    77.  
    78.     @Override
    79.     public InputStream detach() {
    80.         InputStream detachedBuffer = buffer;
    81.         try {
    82.             buffer.reset();
    83.         } catch (IOException e) {
    84.             throw new RuntimeException(e);
    85.         }
    86.         return new BufferInputStream(detachedBuffer);
    87.     }
    88.  
    89.     @Override
    90.     public void close() throws IOException {
    91.         buffer.close();
    92.     }
    93. }
    94.  
    95. package org.springframework.http.converter.protobuf;
    96.  
    97. import com.google.protobuf.ByteOutput;
    98.  
    99. import java.io.ByteArrayOutputStream;
    100. import java.io.IOException;
    101. import org.springframework.http.HttpHeaders;
    102.  
    103. import java.io.ObjectOutputStream;
    104. import java.nio.charset.StandardCharsets;
    105.  
    106. public class MessageOutputStreamUtils {
    107.  
    108.     public static void writePrefix(ByteOutput outputStream, byte[] bytes) throws IOException {
    109.         outputStream.write(bytes, 0, bytes.length);
    110.     }
    111.  
    112.     public static void writeSuffix(ByteOutput outputStream) throws IOException {
    113.         HttpHeaders headers = new HttpHeaders();
    114.         headers.add("grpc-status", "0");
    115.         byte[] bytes;
    116.         try (ByteArrayOutputStream bo = new ByteArrayOutputStream();
    117.              ObjectOutputStream oo = new ObjectOutputStream(bo);) {
    118.             oo.writeObject(headers);
    119.             bytes = bo.toByteArray();
    120.             outputStream.write(bytes, 0, bytes.length);
    121.         }
    122.     }
    123. }

    笔者试着在输出流多输出header，但是传到body里面了，而不是一个返回发送2次header

     

    待解决的问题

    因为缺少最后这个Header，所以grpc原生client端解析结束不正确 

    总结

    grpc实际上本质还是Http2.0+谷歌定制的protobuf，表现形式为rpc调用，依赖变重，如果还需要ssl/tls就会需要证书加密传输，在内部环境实际上是没必要的，适合对外接口和非浏览器模式，可以实现推送（HTTP2.0的能力，现在不推荐用这个能力了），实际上也可以跟传统的Tomcat通信，笔者已经实现调用通过，只有最后的传输结束还没处理好。

  • 相关阅读:
    SSM SpringBoot vue快递柜管理系统
    【信号去噪】基于快速子带自适应滤波 (FSAF)实现信号去噪处理附matlab代码
    元数据简析：定义及管理
    多关键字dp，P1687 机器人小Q
    Springboot+vue的企业人事管理系统（有报告），Javaee项目，springboot vue前后端分离项目。
    Elasticsearch 8.X 如何生成 TB 级的测试数据 ?
    如何写一个sh脚本将一个本地文件通过 scp命令上传到远程的 centos服务器？
    JVM对象创建、对象内存分配、对象内存回收机制
    妈妈的爱，从口腔微生物开始
    销售运营管理
  • 原文地址：https://blog.csdn.net/fenglllle/article/details/127829481