grpc的streamObserver理解以及流式的理解

系统管理员 2022-10-07 14:54 176阅读 0赞

流式的理解参考：grpc-java-demo  
操作端操作流，获取端普通获取即可  
普通的可以用监听者工具类回调通知，也可不用，但是流式的要用监听者方式获取  
被观察者将数据写入即grpc会回调观察者（抽象方法），观察者要定义回调的具体方法  
被观察者即数据的输入方，观察者数据的到达方（接收方），数据的接收方写具体的回调函数方法--具体处理逻辑  
接收的数据  
 responseObserver.onNext(rsp ); 调用回调的方法以及传入的参数  
 requestObserver.onCompleted();  所有回调完成的回调方法，这个在接受方也可写具体的方法逻辑

接受方写好回调返回，传入，发送发回调即可  
含有客户端流的就需要将请求参数返回 request fun(response)  
只有客户端流  
服务端写request回调中，onplement返回（最后返回），客户端可以回调接受  
既有客户端也有服务端的服务端写回调，onnext的同时就返回（伴随着客户端的写，多次返回），客户端回调接受

只有服务端流的返回void fun(request ，response)  
只有服务端流的用迭代获取返回，或者用回调方式  
Iterator<AccountProto.QueryResponse> rspIterator = stub.queryServerStreaming(request);  
        while (rspIterator.hasNext()) \{  
            AccountProto.QueryResponse rsp = rspIterator.next();  
            this.logger.info("QueryResponse: \{\}", rsp);  
        \}  
没有流的服务端可以写成普通的，也可用回调方式  
客户端都只需直接获取  
 AccountProto.QueryResponse rsp = stub.query(request);

即流的接受可以用迭代，也可用回调方式接受，对于迭代只有一次的流可以直接普通接受

客户端流  
服务方：--只需要写接受的参数回调-- 用return 的方式通过方法中的参数类型区别是什么的回调具体实现，写好后返回传出  
/\*\*  
     \* \{@inheritDoc\}  
     \*  
     \* @since 2019-07-16 10:26 Rename: queryClientStreaming -> addClientStreaming  
     \* @since 2019-07-16 10:32 修改参数为 AddRequest、AddResponse  
     \* @since 2019-07-20 11:06 服务端的观察者使用 \{@link ServerCallStreamObserver\}  
     \*/  
    @Override  
    public StreamObserver<AccountProto.AddRequest> addClientStreaming(  
            StreamObserver<AccountProto.AddResponse> responseObserver) \{

// 返回 observer 应对多个请求对象  
        return new ServerCallStreamObserver<>() \{

private final AtomicInteger count = new AtomicInteger(0);

private final ThreadLocal<String> localSerialNo = new InheritableThreadLocal<>();

@Override  
            public void onNext(AccountProto.AddRequest request) \{  
                logger.info("AddRequest: \{\}", request);  
                this.localSerialNo.set(request.getSerialNo());  
                // 获取客户端推送的流数据，处理成功（比如添加到数据库中）后返回结果  
                this.count.incrementAndGet();  
            \}

@Override  
            public void onError(Throwable t) \{  
                // other side closed with non OK  
                logger.error("Build response error", t);  
            \}

@Override  
            public void onCompleted() \{  
                // other side closed with OK  
                AccountProto.AddResponse rsp = AccountProto.AddResponse.newBuilder()  
                        .setCode(Code.OK)  
                        .setSerialNo(this.localSerialNo.get())  
                        .setMsg("All users added.")  
                        .setCount(this.count.get())  
                        .build();  
                responseObserver.onNext(rsp);  
                responseObserver.onCompleted();  
            \}

@Override  
            public boolean isReady() \{  
                return false;  
            \}

@Override  
            public void setOnReadyHandler(Runnable onReadyHandler) \{  
                logger.info("setOnReadyHandler");  
            \}

@Override  
            public void disableAutoInboundFlowControl() \{  
                logger.info("disableAutoInboundFlowControl");  
            \}

@Override  
            public void request(int count) \{  
                logger.info("request: " + count);  
            \}

@Override  
            public void setMessageCompression(boolean enable) \{  
                logger.info("setMessageCompression: " + enable);  
            \}

@Override  
            public boolean isCancelled() \{  
                return false;  
            \}

@Override  
            public void setOnCancelHandler(Runnable onCancelHandler) \{  
                logger.info("setOnCancelHandler");  
            \}

@Override  
            public void setCompression(String compression) \{  
                logger.info("setCompression: " + compression);  
            \}  
        \};  
    \}  
      
      
    客户端流：  
    客户方 ---只需要写接受的参数--返回值回调，写好后参数传出  
    @Test  
    public void testAddClientStreaming04() \{  
        ManagedChannel channel = ManagedChannelBuilder.forAddress("localhost", 8883)  
                .usePlaintext()  
                .build();  
        // 异步 stub  
        AccountServiceGrpc.AccountServiceStub asyncStub = AccountServiceGrpc.newStub(channel);

// client stream  
        logger.info("---client stream rpc---");  
        StreamObserver<AccountProto.AddResponse> responseObserver = new ClientCallStreamObserver<>() \{

@Override  
            public void onNext(AccountProto.AddResponse rsp) \{  
                logger.info("AddResponse: \{\}", rsp);  
            \}

@Override  
            public void onError(Throwable t) \{  
                logger.error("Observe response error", t);  
            \}

@Override  
            public void onCompleted() \{  
                // 关闭channel  
                channel.shutdown();  
            \}

@Override  
            public boolean isReady() \{  
                return false;  
            \}

@Override  
            public void setOnReadyHandler(Runnable onReadyHandler) \{  
                logger.info("setOnReadyHandler");  
            \}

@Override  
            public void disableAutoInboundFlowControl() \{  
                logger.info("disableAutoInboundFlowControl");  
            \}

@Override  
            public void request(int count) \{  
                logger.info("request: \{\}", count);  
            \}

@Override  
            public void setMessageCompression(boolean enable) \{  
                logger.info("setMessageCompression: \{\}", enable);  
            \}

@Override  
            public void cancel(@Nullable String message, @Nullable Throwable cause) \{  
                logger.warn("cancel: " + message, cause);  
            \}  
        \};

StreamObserver<AccountProto.AddRequest> requestObserver = asyncStub.addClientStreaming(responseObserver);  
        AccountProto.AddRequest.Builder builder = AccountProto.AddRequest.newBuilder();  
        for (int i = 1; i <= 1024; i++) \{  
            AccountProto.AddRequest request = builder.setSerialNo("Serial\#" + i)  
                    .setUserId("user\#" + i)  
                    .setUserName("User\#" + i)  
                    .setAmount(this.calcAmount(666.6604D, i))  
                    .build();  
            requestObserver.onNext(request);  
        \}  
        requestObserver.onCompleted();  
        try \{  
            // 线程休眠，等待异步结果  
            Thread.sleep(5000);  
        \} catch (InterruptedException e) \{  
            e.printStackTrace();  
        \}  
    \}  
      
      
  
服务端：  
@Override  
    public StreamObserver<AccountProto.QueryRequest> queryBidiStreaming(  
            StreamObserver<AccountProto.QueryResponse> responseObserver) \{

// 返回observer应对多个请求对象  
        return new ServerCallStreamObserver<>() \{

private final AccountProto.QueryResponse.Builder builder = AccountProto.QueryResponse.newBuilder();

@Override  
            public void onNext(AccountProto.QueryRequest request) \{  
                logger.info("QueryRequest: \{\}", request);  
                for (int i = 1; i <= DATA\_COUNT; i++) \{  
                    AccountProto.QueryResponse rsp = this.builder.setSerialNo(request.getSerialNo() + '\#' + i)  
                            .setCode(Code.OK)  
                            .setMsg("OK; User: " + request.getUserId())  
                            .setAmount(calcAmount(666.6603D, i))  
                            .build();  
                    responseObserver.onNext(rsp);  
                \}  
            \}

@Override  
            public void onCompleted() \{  
                // other side closed with OK  
                responseObserver.onCompleted();  
            \}

@Override  
            public boolean isReady() \{  
                return false;  
            \}

@Override  
            public void setOnReadyHandler(Runnable onReadyHandler) \{  
                logger.info("setOnReadyHandler");  
            \}

@Override  
            public void disableAutoInboundFlowControl() \{  
                logger.info("disableAutoInboundFlowControl");  
            \}

@Override  
            public void request(int count) \{  
                logger.info("request: " + count);  
            \}

@Override  
            public void setMessageCompression(boolean enable) \{  
                logger.info("setMessageCompression: " + enable);  
            \}

@Override  
            public boolean isCancelled() \{  
                return false;  
            \}

@Override  
            public void setOnCancelHandler(Runnable onCancelHandler) \{  
                logger.info("setOnCancelHandler");  
            \}

@Override  
            public void setCompression(String compression) \{  
                logger.info("setCompression: " + compression);  
            \}  
        \};  
    \}

参考：

[https://blog.csdn.net/weixin\_43430036/article/details/101531032][https_blog.csdn.net_weixin_43430036_article_details_101531032]

[https_blog.csdn.net_weixin_43430036_article_details_101531032]: https://blog.csdn.net/weixin_43430036/article/details/101531032