如何捕获非阻塞双向流式gRPC调用中的StatusRuntimeException

在非阻塞gRPC调用中，错误不会像同步调用那样直接抛出到调用线程，而是通过你传入的StreamObserver的onError方法回调通知。你的代码中省略了onError的实现，导致无法捕获这类异常。

修正方案：实现StreamObserver的onError方法 #

修改代码，完整实现StreamObserver的三个核心方法，在onError中处理StatusRuntimeException：

this.executor.submit(() -> {
    try {
        this.toServer = kvStore.connect(new StreamObserver<KVMessage>() {

            @Override
            public void onNext(KVMessage resp) {
                lock.lock();
                response = resp;
                responseReceived.signalAll();
                lock.unlock();
            }

            @Override
            public void onError(Throwable t) {
                // 在这里捕获并处理gRPC调用产生的异常
                if (t instanceof StatusRuntimeException) {
                    StatusRuntimeException e = (StatusRuntimeException) t;
                    System.err.println("*** 连接服务器失败！*** 错误码：" + e.getStatus().getCode() + "，描述：" + e.getStatus().getDescription());
                    // 同步处理逻辑，避免死锁
                    lock.lock();
                    responseReceived.signalAll();
                    lock.unlock();
                }
            }

            @Override
            public void onCompleted() {
                // 调用正常完成时的收尾逻辑
                lock.lock();
                responseReceived.signalAll();
                lock.unlock();
            }
        });
        Thread.sleep(5000);
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        System.err.println("线程被中断");
    }
    gate.countDown();
});

关键说明 #

• StatusRuntimeException会被gRPC框架包装后传递到onError方法，需在此处做类型转换并处理。
• 必须在onError和onCompleted中处理同步逻辑（如释放锁、唤醒等待线程），防止程序陷入死锁。
• 若需监听通道级别的全局错误（如DNS解析失败导致通道无法建立），可给ManagedChannel添加状态监听器：

channel.notifyWhenStateChanged(ConnectivityState.READY, () -> {
    ConnectivityState currentState = channel.getState(false);
    if (currentState == ConnectivityState.TRANSIENT_FAILURE) {
        Status status = channel.getStateInfo(currentState).getStatus();
        System.err.println("通道状态异常：" + status.getCode() + "，描述：" + status.getDescription());
    }
});

内容的提问来源于stack exchange，提问作者nawK