grpc中的Java多线程

Question

我有一个关于JAVA多线程的问题。

我有一个jetty webapp与grpc-流式传输-客户端。一切都很好，但我如何建立一个模型来获取流式传输数据？

webapp是用jsf构建的。因为我有一个控制器，它调用一个处理程序类来启动流：

    public void startStream(){
    if(streamHandler!=null & activatedStream == false){
        streamHandler.startStreamClient();          
        activatedStream = true;
    }else{
        FacesMessage message = new FacesMessage(FacesMessage.SEVERITY_ERROR, 
                "Could not initialize the StreamHandler and Client Class or a Stream still runs. Please check the logs.", "Stream is running: "+String.valueOf(activatedStream));
        FacesContext.getCurrentInstance().addMessage(null, message);
    }
}

此方法简单地启动客户端和流。

    public void startStreamClient(){
    log.info("Calling startMethod of Handler............");
    CountDownLatch finishLatch;
    if(this.client.isChannelShutdown()& this.client!=null){

        this.client=new StreamClient(this.serverHost, this.serverPort);

        try{
            finishLatch = this.client.imageStream(this.startRequest);
        }catch(Exception e){
            log.warn("Error while starting the imageStream: "+e.getLocalizedMessage(), e);
        }
    }else{
        finishLatch = client.imageStream(this.startRequest);
    }
}

检查倒计时锁存器的实现仍然缺失。但在这种情况下，这并不重要。

响应如下：onNext（）-方法提供流式数据：

public CountDownLatch imageStream(StreamRequest request){

    log.info("Calling imageStream-asnychStub...............");

    CountDownLatch finish = new CountDownLatch(1);

    /**
     *  The asyncStub is calling the rpc-Function with a new StreamObserver for the given Responses from the Server.
     */
    StreamObserver<StreamRequest> requestOberserver = asyncStub.streamImagaData(new StreamObserver<StreamResponse>() {

        /**
         * The onNext Method is getting the imageDate, if it is send
         */
        @Override
        public void onNext(StreamResponse response) {
            System.out.println("Data-Input: "+response.getImageData().length());
        }

        /**
         * The onError Method is getting an Exception Object if it is thrown
         */
        @Override
        public void onError(Throwable t) {
            log.warn("Bidirectional Stream with Server an Client: "+t.getLocalizedMessage(), t);
        }           

        /**
         * The onCompleted is for ending the Stream and reduces the CountDownLatch by one
         */
        @Override
        public void onCompleted() {
            log.info("Bidirectional Streaming has finished....");
            finish.countDown();
        }
    });

    /**
     * This Block is for sending a StreamRequest to the Server.
     */
    try{
        log.info("Sending a Streaming-Request to Server with State: "+request.getStreamState().name());
        requestOberserver.onNext(request);
    }catch (RuntimeException ex) {
        log.warn("Error sending requst to Server: "+ex.getLocalizedMessage(), ex);
        requestOberserver.onError(ex);
    }
    requestOberserver.onCompleted();

    return finish;
}

图像数据简单地打印在屏幕上。我试图建立一个消费者-生产者模型，但失败了，因为响应以StreamObserver的内部类型返回。

如何实时获取这些数据。我必须创建StreamObserver的正式实现吗？或者我必须在哪里放置额外的线程？线程是唯一的选择吗？我需要打电话吗？

提前感谢。

Answer 1

我通过实现观察者模式解决了这个问题。

ManagedBean成为StreamClient的观察者。