gRPC

 GRPC是google開源的一個高性能、跨語言的RPC框架，基於HTTP2協議，基於protobuf 3.x，基於Netty 4.x +。GRPC與thrift、avro-rpc等其實在總體原理上並沒有太大的區別，簡而言之GRPC並沒有太多突破性的創新。（如下描述，均基於JAVA語言的實現）

    對於開發者而言：

    1）需要使用protobuf定義接口，即.proto文件

    2）然后使用compile工具生成特定語言的執行代碼，比如JAVA、C/C++、Python等。類似於thrift，為了解決跨語言問題。

    3）啟動一個Server端，server端通過偵聽指定的port，來等待Client鏈接請求，通常使用Netty來構建，GRPC內置了Netty的支持。

    4）啟動一個或者多個Client端，Client也是基於Netty，Client通過與Server建立TCP長鏈接，並發送請求；Request與Response均被封裝成HTTP2的stream Frame，通過Netty Channel進行交互。

 

    對於GRPC的“鼓吹”，本文不多表述，截止到今日，GRPC仍然處於開發階段，尚沒有release版本，而且特性也很多需要補充；GRPC基於protobuf 3.x，但是protobuf 3.x也沒有release版本；雖然HTTP2協議已成定局，但尚未被主流web容器包括代理服務器支持，這意味着GRPC在HTTP負載均衡方面尚有欠缺；最終，在短期內我們還不能在production環境中實施，可以做技術儲備。不過GRPC的缺點，在將來將會成為它的優點，我們需要時間等待它的成熟。

    1）GRPC尚未提供連接池

    2）尚未提供“服務發現”、“負載均衡”機制

    3）因為基於HTTP2，絕大部多數HTTP Server、Nginx都尚不支持，即Nginx不能將GRPC請求作為HTTP請求來負載均衡，而是作為普通的TCP請求。（nginx將會在1.9版本支持）

    4）GRPC尚不成熟，易用性還不是很理想；就本人而言，我還是希望GRPC能夠像hessian一樣：無IDL文件，無需代碼生成，接口通過HTTP表達。

    5）Spring容器尚未提供整合。

 

    在實際應用中，GRPC尚未完全提供連接池、服務自動發現、進程內負載均衡等高級特性，需要開發人員額外的封裝；最大的問題，就是GRPC生成的接口，調用方式實在是不太便捷（JAVA），最起碼與thrift相比還有差距，希望未來能夠有所改進。

 

一、實例

    1、proto文件

    GRPC並沒有創造新的序列化協議，而是使用已有的protobuf；基於protobuf來聲明數據模型和RPC接口服務，當然protobuf是一個非常優秀的協議框架。關於protobuf 3.x的相關文檔，請參見【protobuf 3】

    接下來，我們設計一個sayHello接口，我們將數據模型和RPC接口分別保存在兩個文件中。

    1）TestModel.proto

 

syntax = "proto3";
package com.test.grpc;
option java_package = "com.test.grpc.service.model";
message TestRequest{
    string name  = 1;
    int32 id    = 2;
}
message TestResponse{
    string message = 1;
}

 

    2）TestService.proto

syntax = "proto3";
package com.test.grpc;
option java_package = "com.test.grpc.service";
import "TestModel.proto";
service TestRpcService{
    rpc sayHello(TestRequest) returns (TestResponse);
}

 

    proto文件中需要注意加上“syntax”，表示使用protobuf 3的語法。

    2、生成JAVA代碼

    生成代碼，我們最好借助於maven插件，可以在pom文件中增加如下信息：

<pluginRepositories><!-- 插件庫 -->
    <pluginRepository>
        <id>protoc-plugin</id>
        <url>https://dl.bintray.com/sergei-ivanov/maven/</url>
    </pluginRepository>
</pluginRepositories>
<build>
    <extensions>
        <extension>
            <groupId>kr.motd.maven</groupId>
            <artifactId>os-maven-plugin</artifactId>
            <version>1.4.0.Final</version>
        </extension>
    </extensions>
    <plugins>
        <plugin>
            <groupId>com.google.protobuf.tools</groupId>
            <artifactId>maven-protoc-plugin</artifactId>
            <version>0.4.4</version>
            <configuration>
                <protocArtifact>com.google.protobuf:protoc:3.0.0-beta-2:exe:${os.detected.classifier}</protocArtifact>
                <pluginId>grpc-java</pluginId>
                <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
            </configuration>
            <executions>
                <execution>
                    <goals>
                        <goal>compile</goal>
                        <goal>compile-custom</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

 

    然后只需要執行“mvn compile”指令即可，此后我們會在項目的target目錄下看到生成的classes文件，當然最終我們還是需要將service打成jar包發布的。maven仍然可以幫助我們做這些工作，由.proto生成classes是在compile階段，那么jar階段仍然是可以將classes打成jar，只需要借助maven-jar-plugin插件即可。

    3、開發Server端服務（簡例）

//server端實現類，擴展原有接口
public class TestServiceImpl implements TestRpcServiceGrpc.TestRpcService {

    @Override
    public void sayHello(TestModel.TestRequest request, StreamObserver<TestModel.TestResponse> responseObserver) {
        String result = request.getName() + request.getId();
        TestModel.TestResponse response = TestModel.TestResponse.newBuilder().setMessage(result).build();
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}

 

public class TestServer {

    public static void main(String[] args) throws Exception{

        ServerImpl server = NettyServerBuilder.forPort(50010).addService(TestRpcServiceGrpc.bindService(new TestServiceImpl())).build();
        server.start();
        server.awaitTermination();//阻塞直到退出
    }
}

 

    稍后啟動TestServer即可。

    4、開發Client端（簡例）

public class TestClient {

    private final TestRpcServiceGrpc.TestRpcServiceBlockingStub client;
    public TestClient(String host,int port) {
        ManagedChannel channel =  NettyChannelBuilder.forAddress(host, port).usePlaintext(true).build();
        client = TestRpcServiceGrpc.newBlockingStub(channel).withDeadlineAfter(60000, TimeUnit.MILLISECONDS);
    }

    public String sayHello(String name,Integer id) {
        TestModel.TestRequest request = TestModel.TestRequest.newBuilder().setId(id).setName(name).build();
        TestModel.TestResponse response = client.sayHello(request);
        return response.getMessage();
    }
}

 

    然后我們運行即可，代碼非常簡單，當然無論是Client還是Server端，我們還有其他額外的參數可以配置，我們稍后詳細介紹。

 

二、原理解析

    GRPC的Client與Server，均通過Netty Channel作為數據通信，序列化、反序列化則使用Protobuf，每個請求都將被封裝成HTTP2的Stream，在整個生命周期中，客戶端Channel應該保持長連接，而不是每次調用重新創建Channel、響應結束后關閉Channel（即短連接、交互式的RPC），目的就是達到鏈接的復用，進而提高交互效率。

 

    1、Server端

    我們通常使用NettyServerBuilder，即IO處理模型基於Netty，將來可能會支持其他的IO模型。Netty Server的IO模型簡析：

    1）創建ServerBootstrap，設定BossGroup與workerGroup線程池

    2）注冊childHandler，用來處理客戶端鏈接中的請求成幀

    3）bind到指定的port，即內部初始化ServerSocketChannel等，開始偵聽和接受客戶端鏈接。

    4）BossGroup中的線程用於accept客戶端鏈接，並轉發（輪訓）給workerGroup中的線程。

    5）workerGroup中的特定線程用於初始化客戶端鏈接，初始化pipeline和handler，並將其注冊到worker線程的selector上（每個worker線程持有一個selector，不共享）

    6）selector上發生讀寫事件后，獲取事件所屬的鏈接句柄，然后執行handler（inbound），同時進行拆封package，handler執行完畢后，數據寫入通過，由outbound handler處理（封包）通過鏈接發出。    注意每個worker線程上的數據請求是隊列化的。

    參見源碼：SingleThreadEventLoop、NioEventLoop。（請求隊列化）

 

    GRPC而言，只是對Netty Server的簡單封裝，底層使用了PlaintextHandler、Http2ConnectionHandler的相關封裝等。具體Framer、Stream方式請參考Http2相關文檔。

    1）bossEventLoopGroup：如果沒指定，默認為一個static共享的對象，即JVM內所有的NettyServer都使用同一個Group，默認線程池大小為1。

    2）workerEventLoopGroup：如果沒指定，默認為一個static共享的對象，線程池大小為coreSize * 2。這兩個對象采用默認值並不會帶來問題；通常情況下，即使你的application中有多個GRPC Server，默認值也一樣能夠帶來收益。不合適的線程池大小，有可能會是性能受限。

    3）channelType：默認為NioServerSocketChannel，通常我們采用默認值；當然你也可以開發自己的類。如果此值為NioServerSocketChannel，則開啟keepalive，同時設定SO_BACKLOG為128；BACKLOG就是系統底層已經建立引入鏈接但是尚未被accept的Socket隊列的大小，在鏈接密集型（特別是短連接）時，如果隊列超過此值，新的創建鏈接請求將會被拒絕（有可能你在壓力測試時，會遇到這樣的問題），keepalive和BACKLOG特性目前無法直接修改。

[root@sh149 ~]# sysctl -a|grep tcp_keepalive
net.ipv4.tcp_keepalive_time = 60  ##單位：秒
net.ipv4.tcp_keepalive_probes = 9
net.ipv4.tcp_keepalive_intvl = 75 ##單位：秒
##可以在/etc/sysctl.conf查看和修改相關值
##tcp_keepalive_time：最后一個實際數據包發送完畢后，首個keepalive探測包發送的時間。
##如果首個keepalive包探測成功，那么鏈接會被標記為keepalive（首先TCP開啟了keepalive）
##此后此參數將不再生效，而是使用下述的2個參數繼續探測
##tcp_keepalive_intvl:此后，無論通道上是否發生數據交換，keepalive探測包發送的時間間隔
##tcp_keepalive_probes:在斷定鏈接失效之前，嘗試發送探測包的次數；
##如果都失敗，則斷定鏈接已關閉。

 

    對於Server端，我們需要關注上述keepalive的一些設置；如果Netty Client在空閑一段時間后，Server端會主動關閉鏈接，有可能Client仍然保持鏈接的句柄，將會導致RPC調用時發生異常。這也會導致GRPC客戶端調用時偶爾發生錯誤的原因之一。

    4）followControlWindow：流量控制的窗口大小，單位：字節，默認值為1M，HTTP2中的“Flow Control”特性；連接上，已經發送尚未ACK的數據幀大小，比如window大小為100K，且winow已滿，每次向Client發送消息時，如果客戶端反饋ACK（攜帶此次ACK數據的大小），window將會減掉此大小；每次向window中添加亟待發送的數據時，window增加；如果window中的數據已達到限定值，它將不能繼續添加數據，只能等待Client端ACK。

    5）maxConcurrentCallPerConnection：每個connection允許的最大並發請求數，默認值為Integer.MAX_VALUE;如果此連接上已經接受但尚未響應的streams個數達到此值，新的請求將會被拒絕。為了避免TCP通道的過度擁堵，我們可以適度調整此值，以便Server端平穩處理，畢竟buffer太多的streams會對server的內存造成巨大壓力。

    6）maxMessageSize：每次調用允許發送的最大數據量，默認為100M。

    7）maxHeaderListSize：每次調用允許發送的header的最大條數，GRPC中默認為8192。

 

    對於其他的比如SSL/TSL等，可以參考其他文檔。

    GRPC Server端，還有一個最終要的方法：addService。【如下文service代理模式】

    在此之前，我們需要介紹一下bindService方法，每個GRPC生成的service代碼中都有此方法，它以硬編碼的方式遍歷此service的方法列表，將每個方法的調用過程都與“被代理實例”綁定，這個模式有點類似於靜態代理，比如調用sayHello方法時，其實內部直接調用“被代理實例”的sayHello方法（參見MethodHandler.invoke方法，每個方法都有一個唯一的index，通過硬編碼方式執行）；bindService方法的最終目的是創建一個ServerServiceDefinition對象，這個對象內部位置一個map，key為此Service的方法的全名（fullname，{package}.{service}.{method}）,value就是此方法的GRPC封裝類（ServerMethodDefinition）。

    源碼分析：

private static final int METHODID_SAY_HELLO = 0;
private static class MethodHandlers<Req, Resp> implements
      ... {
    private final TestRpcService serviceImpl;//實際被代理實例
    private final int methodId;

    public MethodHandlers(TestRpcService serviceImpl, int methodId) {
      this.serviceImpl = serviceImpl;
      this.methodId = methodId;
    }

    @java.lang.SuppressWarnings("unchecked")
    public void invoke(Req request, io.grpc.stub.StreamObserver<Resp> responseObserver) {
      switch (methodId) {
        case METHODID_SAY_HELLO:        //通過方法的index來判定具體需要代理那個方法
          serviceImpl.sayHello((com.test.grpc.service.model.TestModel.TestRequest) request,
              (io.grpc.stub.StreamObserver<com.test.grpc.service.model.TestModel.TestResponse>) responseObserver);
          break;
        default:
          throw new AssertionError();
      }
    }
    ....
  }

  public static io.grpc.ServerServiceDefinition bindService(
      final TestRpcService serviceImpl) {
    return io.grpc.ServerServiceDefinition.builder(SERVICE_NAME)
        .addMethod(
          METHOD_SAY_HELLO,
          asyncUnaryCall(
            new MethodHandlers<
              com.test.grpc.service.model.TestModel.TestRequest,
              com.test.grpc.service.model.TestModel.TestResponse>(
                serviceImpl, METHODID_SAY_HELLO)))
        .build();
  }

 

    addService方法可以添加多個Service，即一個Netty Server可以為多個service服務，這並不違背設計模式和架構模式。addService方法將會把service保存在內部的一個map中，key為serviceName（即{package}.{service}）,value就是上述bindService生成的對象。

 

    那么究竟Server端是如何解析RPC過程的？Client在調用時會將調用的service名稱 + method信息保存在一個GRPC“保留”的header中，那么Server端即可通過獲取這個特定的header信息，就可以得知此stream需要請求的service、以及其method，那么接下來只需要從上述提到的map中找到service，然后找到此method，直接代理調用即可。執行結果在Encoder之后發送給Client。（參見：NettyServerHandler）

 

    因為是map存儲，所以我們需要在定義.proto文件時，盡可能的指定package信息，以避免因為service過多導致名稱可能重復的問題。

 

    2、Client端

    我們使用ManagedChannelBuilder來創建客戶端channel，ManagedChannelBuilder使用了provider機制，具體是創建了哪種channel有provider決定，可以參看META-INF下同類名的文件中的注冊信息。當前Channel有2種：NettyChannelBuilder與OkHttpChannelBuilder。本人的當前版本中為NettyChannelBuilder；我們可以直接使用NettyChannelBuilder來構建channel。如下描述則針對NettyChannelBuilder：

    配置參數與NettyServerBuilder基本類似，再次不再贅言。默認情況下，Client端默認的eventLoopGroup線程池也是static的，全局共享的，默認線程個數為coreSize * 2。合理的線程池個數可以提高客戶端的吞吐能力。

 

    ManagedChannel是客戶端最核心的類，它表示邏輯上的一個channel；底層持有一個物理的transport（TCP通道，參見NettyClientTransport），並負責維護此transport的活性；即在RPC調用的任何時機，如果檢測到底層transport處於關閉狀態（terminated），將會嘗試重建transport。（參見TransportSet.obtainActiveTransport()）

    通常情況下，我們不需要在RPC調用結束后就關閉Channel，Channel可以被一直重用，直到Client不再需要請求位置或者Channel無法真的異常中斷而無法繼續使用。當然，為了提高Client端application的整體並發能力，我們可以使用連接池模式，即創建多個ManagedChannel，然后使用輪訓、隨機等算法，在每次RPC請求時選擇一個Channel即可。（備注，連接池特性，目前GRPC尚未提供，需要額外的開發）

 

    每個Service客戶端，都生成了2種stub：BlockingStub和FutureStub；這兩個Stub內部調用過程幾乎一樣，唯一不同的是BlockingStub的方法直接返回Response Model，而FutureStub返回一個Future對象。BlockingStub內部也是基於Future機制，只是封裝了阻塞等待的過程：

try {
        //也是基於Future
      ListenableFuture<RespT> responseFuture = futureUnaryCall(call, param);
      //阻塞過程
      while (!responseFuture.isDone()) {
        try {
          executor.waitAndDrain();
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          throw Status.CANCELLED.withCause(e).asRuntimeException();
        }
      }
      return getUnchecked(responseFuture);
    } catch (Throwable t) {
      call.cancel();
      throw t instanceof RuntimeException ? (RuntimeException) t : new RuntimeException(t);
}

 

    創建一個Stub的成本是非常低的，我們可以在每次請求時都通過channel創建新的stub，這並不會帶來任何問題（只不過是創建了大量對象）；其實更好的方式是，我們應該使用一個Stub發送多次請求，即Stub也是可以重用的；直到Stub上的狀態異常而無法使用。最常見的異常，就是“io.grpc.StatusRuntimeException: DEADLINE_EXCEEDED”，即表示DEADLINE時間過期，我們可以為每個Stub配置deadline時間，那么如果此stub被使用的時長超過此值（不是空閑的時間），將不能再發送請求，此時我們應該創建新的Stub。很多人想盡辦法來使用“withDeadlineAfter”方法來實現一些奇怪的事情，此參數的主要目的就是表明：此stub只能被使用X時長，此后將不能再進行請求，應該被釋放。所以，它並不能實現類似於“keepAlive”的語義，即使我們需要keepAlive，也應該在Channel級別，而不是在一個Stub上。

 

    如果你使用了連接池，那么其實連接池不應該關注DEADLINE的錯誤，只要Channel本身沒有terminated即可；就把這個問題交給調用者處理。如果你也對Stub使用了對象池，那么你就可能需要關注這個情況了，你不應該向調用者返回一個“DEADLINE”的stub，或者如果調用者發現了DEADLINE，你的對象池應該能夠移除它。

 

    1）實例化ManagedChannel，此channel可以被任意多個Stub實例引用；如上文說述，我們可以通過創建Channel池，來提高application整體的吞吐能力。此Channel實例，不應該被shutdown，直到Client端停止服務；在任何時候，特別是創建Stub時，我們應該判定Channel的狀態。

synchronized (this) {
    if (channel.isShutdown() || channel.isTerminated()) {
        channel = ManagedChannelBuilder.forAddress(poolConfig.host, poolConfig.port).usePlaintext(true).build();
    }
    //new Stub
}

//或者
ManagedChannel channel = (ManagedChannel)client.getChannel();
if(channel.isShutdown() || channel.isTerminated()) {
    client = createBlockStub();
}
client.sayHello(...)

 

    因為Channel是可以多路復用，所以我們用Pool機制（比如commons-pool）也可以實現連接池，只是這種池並非完全符合GRPC/HTTP2的設計語義，因為GRPC允許一個Channel上連續發送對個Requests（然后一次性接收多個Responses），而不是“交互式”的Request-Response模式，當然這么使用並不會有任何問題。

 

    2）對於批量調用的場景，我們可以使用FutureStub，對於普通的業務類型RPC，我們應該使用BlockingStub。

    3）每個RPC方法的調用，比如sayHello，調用開始后，將會為每個調用請求創建一個ClientCall實例，其內部封裝了調用的方法、配置選項（headers）等。此后將會創建Stream對象，每個Stream都持有唯一的streamId，它是Transport用於分揀Response的憑證。最終調用的所有參數都會被封裝在Stream中。

    4）檢測DEADLINE，是否已經過期，如果過期，將使用FailingClientStream對象來模擬整個RPC過程，當然請求不會通過通道發出，直接經過異常流處理過程。

    5）然后獲取transport，如果此時檢測到transport已經中斷，則重建transport。（自動重練機制，ClientCallImpl.start()方法）

    6）發送請求參數，即我們Request實例。一次RPC調用，數據是分多次發送，但是ClientCall在創建時已經綁定到了指定的線程上，所以數據發送總是通過一個線程進行（不會亂序）。

    7）將ClientCall實例置為halfClose，即半關閉，並不是將底層Channel或者Transport半關閉，只是邏輯上限定此ClientCall實例上將不能繼續發送任何stream信息，而是等待Response。

    8）Netty底層IO將會對reponse數據流進行解包（Http2ConnectionDecoder）,並根據streamId分揀Response，同時喚醒響應的ClientCalls阻塞。（參見ClientCalls，GrpcFuture）

    9）如果是BlockingStub，則請求返回，如果響應中包含應用異常，則封裝后拋出；如果是網絡異常，則可能觸發Channel重建、Stream重置等。

 

 

    到此為止，已經把GRPC的基本原理描述完畢，此后如果有其他問題，則繼續補充