grpc錯誤處理

對於rpc調用異常處理是需要考慮的一個方面，這里介紹一下grpc中對異常的處理。注：這里的案例都是改造了之前的接口

簡單

簡單模式是一個request對應一個response。

//服務端
public void simpleHello(ProtoObj.Person request,
                  StreamObserver<ProtoObj.Result> responseObserver) {
    System.out.println(request.getMyName()+" calling");
	//返回一個包裝成Exception的Status來返回錯誤信息，如果直接使用Throwable，客戶端無法獲得錯誤信息
    responseObserver.onError(Status.INTERNAL.withDescription("error desc").asRuntimeException());
    //如果調用了onError會自動complete無需手動complete
	//responseObserver.onCompleted();
}
//客戶端
@Test
public void  simple() throws InterruptedException {
    final ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 8080).usePlaintext(true).build();
    HelloServiceGrpc.HelloServiceBlockingStub blockingStub = HelloServiceGrpc.newBlockingStub(channel);

    ProtoObj.Person person = ProtoObj.Person.newBuilder().setMyName("World").build();
    //simple
    System.out.println("---simple rpc---");
    try {
		//try catch遠程調用
        System.out.println(blockingStub.simpleHello(person).getString());
    }catch(Exception e){
		//將異常轉換為status可以得到對應的異常信息
        Status status = Status.fromThrowable(e);
        status.asException().printStackTrace();
    }
    channel.shutdown();
}
//輸出
---simple rpc---
io.grpc.StatusException: INTERNAL: error desc
	at io.grpc.Status.asException(Status.java:548)
	at blog.HelloClient.simple(HelloClient.java:100)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	...

客戶端流式

客戶端流式是多個request對應單個response。

//服務端
@Override
public StreamObserver<ProtoObj.Person> clientStreamHello(
       final StreamObserver<ProtoObj.Result> responseObserver) {
   return new StreamObserver<ProtoObj.Person>(){
       private ProtoObj.Result.Builder builder=ProtoObj.Result.newBuilder();
       @Override
       public void onNext(ProtoObj.Person value) {
           builder.setString(builder.getString() +"," + value.getMyName());
       }

       @Override
       public void onError(Throwable t) {
           responseObserver.onError(new Exception("custom error"));
       }

       @Override
       public void onCompleted() {
           builder.setString("hello"+builder.getString());
		   //返回異常
           responseObserver.onError(Status.INTERNAL.withDescription("error desc").asRuntimeException());
           //responseObserver.onNext(builder.build());
           //responseObserver.onCompleted();
       }
   };
}
//客戶端
@Test
public void client() throws InterruptedException {

    final ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 8080).usePlaintext(true).build();
    HelloServiceGrpc.HelloServiceStub asyncStub = HelloServiceGrpc.newStub(channel);

    ProtoObj.Person person = ProtoObj.Person.newBuilder().setMyName("World").build();
	//使用latch來判斷是否結束
    final CountDownLatch latch = new CountDownLatch(1);

    //client side
    System.out.println("---client stream rpc---");
    StreamObserver<ProtoObj.Result> responseObserver = new StreamObserver<ProtoObj.Result>() {
        @Override
        public void onNext(ProtoObj.Result result) {
            System.out.println("client stream--" + result.getString());
        }

        @Override
        public void onError(Throwable t) {
            latch.countDown();
			//處理異常
            Status status = Status.fromThrowable(t);
            status.asException().printStackTrace();
        }

        @Override
        public void onCompleted() {
            latch.countDown();
        }
    };
	//發送請求
    StreamObserver<ProtoObj.Person> clientStreamObserver = asyncStub.clientStreamHello(responseObserver);
    clientStreamObserver.onNext(ProtoObj.Person.newBuilder().setMyName("World").build());
    clientStreamObserver.onNext(ProtoObj.Person.newBuilder().setMyName("World2").build());
    clientStreamObserver.onCompleted();
	//使用latch判斷是否結束調用
    if (!Uninterruptibles.awaitUninterruptibly(latch, 1, TimeUnit.SECONDS)) {
        throw new RuntimeException("timeout!");
    }
    channel.shutdown();

}
//輸出
---client stream rpc---
io.grpc.StatusException: INTERNAL: error desc
	at io.grpc.Status.asException(Status.java:548)
	at blog.HelloClient$3.onError(HelloClient.java:148)
	at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:392)

服務端流式

服務端流式是單個request對應多個response。

//服務端
@Override
public void serverStreamHello(ProtoObj.Person request,
                        StreamObserver<ProtoObj.Result> responseObserver) {
    System.out.println(request.getMyName()+" calling");
    responseObserver.onNext(ProtoObj.Result.newBuilder().setString("hello, "+request.getMyName()).build());
    responseObserver.onError(Status.INTERNAL.withDescription("error desc").asRuntimeException());
    //error之后就不能繼續調用next了
	//responseObserver.onNext(ProtoObj.Result.newBuilder().setString("hello3, "+request.getMyName()).build());
    //responseObserver.onCompleted();
}
//客戶端
    @Test
    public void server(){
        final ManagedChannel channel = ManagedChannelBuilder.forAddress("127.0.0.1", 8080).usePlaintext(true).build();
        HelloServiceGrpc.HelloServiceBlockingStub blockingStub = HelloServiceGrpc.newBlockingStub(channel);

        ProtoObj.Person person = ProtoObj.Person.newBuilder().setMyName("World").build();

        //server side
        System.out.println("---server stream rpc---");
        Iterator<ProtoObj.Result> it = blockingStub.serverStreamHello(person);
        try{
            while (it.hasNext()) {
                System.out.print(it.next());
            }
        }catch (Exception e){
            Status status = Status.fromThrowable(e);
            status.asException().printStackTrace();
        }
        channel.shutdown();
    }
//輸出
---server stream rpc---
string: "hello, World"
io.grpc.StatusException: INTERNAL: error desc
	at io.grpc.Status.asException(Status.java:548)
	at blog.HelloClient.server(HelloClient.java:122)
	...

雙向流式

雙向流式是多個request對應多個response，異常處理和客戶端流式類似（使用異步的newStub），這里就不再贅述。