Vertx GRPC 流观察器答案

【问题标题】：Vertx GRPC Stream ObserverVertx GRPC 流观察器
【发布时间】：2022-01-20 17:57:27
【问题描述】：

我已经使用 vertx 实现了 GRPC 双向流服务。一旦我从服务器端得到响应，我必须根据响应执行另一个过程。为此，我使用了requestStreamObserver.onCompleted()，但是在此调用之后，我得到了

流已经完成，不允许再调用。

从服务器端。

所以我的问题是如何在不终止服务器流的情况下读取响应流并执行该过程

原型文件

syntax = "prto3";

service userService{
    rpc getData(stream Request) returns(stream Response);
}

message Request{
    int32 id = 1;
}

message Response{
    int32 responseCode = 1;
    string username = 2;
}

客户端实现

public void bidi() {

    List<String> responseList = new ArrayList<>();

    final CountDownLatch finishLatch = new CountDownLatch(1);
    StreamObserver<Response> responseStreamObserver = new StreamObserver<Response>() {
        @Override
        public void onNext(Response response) {
            System.out.println(response.getResponseCode() + " - " + response.getUsername());
            responseList.add(response.getUsername());
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("Error " + throwable.getMessage());
            finishLatch.countDown();
        }

        @Override
        public void onCompleted() {
            System.out.println("completed stream");
            finishLatch.countDown();
        }
    };

    StreamObserver<Request> requestStreamObserver = nonBlockingStub.requestData(responseStreamObserver);

    try {
        for (int i = 1; i < 6; i++) {
            Request request = Request.newBuilder().setId(i).build();
            requestStreamObserver.onNext(request);
            if (finishLatch.getCount() == 0) {
                return;
            }
        }            
    } catch (Exception e) {
        e.printStackTrace();
    }

    requestStreamObserver.onCompleted();
    try {
        finishLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    //These code blocks need to be executed after completing the response
    for(String data:responseList){
        System.out.println(data);
    }
    doStuff();
}

由于我无权访问服务器端，我无法发布服务器端实现，但请求正在接受一个流并从服务器返回一个流。

谢谢。

【问题讨论】：

标签： java grpc vert.x

【解决方案1】：

所以这就是我解决这个问题的方法。

使用AtomicInteger 并在发出请求时增加计数。然后作为响应，观察者必须在收到响应后递减计数器。如果计数器为 0，则调用 CountdownLatch 的 countDown() 并执行其他操作。

    public void bidi() {

    List<String> responseList = new ArrayList<>();
    AtomicInteger counter = new AtomicInteger();

    final CountDownLatch finishLatch = new CountDownLatch(1);
    StreamObserver<Response> responseStreamObserver = new StreamObserver<Response>() {
        @Override
        public void onNext(Response response) {
            System.out.println(response.getResponseCode() + " - " + response.getUsername());
            responseList.add(response.getUsername());
            counter.decrementAndGet(); //decrement the counter
            if(counter.get() == 0){
                finishLatch.countDown();
            }
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("Error " + throwable.getMessage());
            finishLatch.countDown();
        }

        @Override
        public void onCompleted() {
            System.out.println("completed stream");
            finishLatch.countDown();
        }
    };

    StreamObserver<Request> requestStreamObserver = nonBlockingStub.requestData(responseStreamObserver);

    try {
        for (int i = 1; i < 6; i++) {
            Request request = Request.newBuilder().setId(i).build();
            requestStreamObserver.onNext(request);
            counter.incrementAndGet();//increment the counter
            if (finishLatch.getCount() == 0) {
                return;
            }
        }            
    } catch (Exception e) {
        e.printStackTrace();
    }

    try {
        finishLatch.await();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    
    for(String data:responseList){
        System.out.println(data);
    }
    doStuff();
}

【讨论】：