gRPC双向流服务500客户端并发运行时异常求助

核心问题诊断 #

从异常堆栈和代码实现来看，问题根源在于线程安全缺失和流式调用生命周期管理不当：

1. 共享集合clientObservers、clientRequestCounts未使用线程安全实现，并发场景下触发竞态条件
2. 向客户端推送响应时未检查流的活跃状态，导致向已关闭/取消的流写入数据
3. 请求计数的更新操作非原子，引发计数丢失或错误
4. 未处理客户端主动取消调用的场景，导致写入操作抛出异常

具体修复步骤 #

1. 替换为线程安全集合 #

将全局共享的集合改为线程安全实现，避免并发读写冲突：

// 替换HashMap为ConcurrentHashMap，保证线程安全
private final ConcurrentHashMap<String, StreamObserver<DataResponse>> clientObservers = new ConcurrentHashMap<>();
// 使用AtomicInteger封装计数，确保原子更新
private final ConcurrentHashMap<String, AtomicInteger> clientRequestCounts = new ConcurrentHashMap<>();

2. 原子化更新请求计数 #

利用AtomicInteger的原子操作替代手动get+put，避免计数丢失：

// 替换原有的计数更新逻辑
int currentCount = clientRequestCounts.computeIfAbsent(clientId, k -> new AtomicInteger())
                                     .incrementAndGet();

3. 检查流状态并处理取消事件 #

将responseObserver转换为ServerCallStreamObserver，添加取消处理器并在发送响应前检查流是否活跃：

@Override
public void onNext(DataRequest request) {
    clientId = request.getClientId();
    ServerCallStreamObserver<DataResponse> serverObserver = (ServerCallStreamObserver<DataResponse>) responseObserver;
    
    if (isFirstRequest) {
        // 设置取消处理器，客户端取消时自动清理资源
        serverObserver.setOnCancelHandler(() -> {
            clientObservers.remove(clientId);
            clientRequestCounts.remove(clientId);
        });
        // 使用putIfAbsent避免覆盖已有连接（处理客户端重连场景）
        clientObservers.putIfAbsent(clientId, serverObserver);
        isFirstRequest = false;
    }
    
    // 原子更新请求计数
    int currentCount = clientRequestCounts.computeIfAbsent(clientId, k -> new AtomicInteger())
                                         .incrementAndGet();
    
    processChunk(request.getName());
    
    // 仅当流未取消时发送响应
    if (currentCount % 5 == 0 && !serverObserver.isCancelled()) {
        DataResponse response = DataResponse.newBuilder()
                .setMessage("****Notification sent****")
                .build();
        serverObserver.onNext(response);
    }
}

4. 安全处理流结束与错误场景 #

在onCompleted和onError中清理资源，避免无效引用：

@Override
public void onCompleted() {
    System.out.println("Bidirectional streaming RPC completed");
    StreamObserver<DataResponse> observer = clientObservers.get(clientId);
    if (observer != null && !((ServerCallStreamObserver<DataResponse>) observer).isCancelled()) {
        observer.onCompleted();
    }
    // 清理客户端关联资源
    clientObservers.remove(clientId);
    clientRequestCounts.remove(clientId);
}

@Override
public void onError(Throwable throwable) {
    System.err.println("Encountered error in bidirectional stream: " + throwable);
    // 异常时立即清理资源
    clientObservers.remove(clientId);
    clientRequestCounts.remove(clientId);
}

额外优化建议 #

• 为gRPC连接设置超时时间，避免无效连接长期占用资源
• 定期扫描clientObservers，清理已取消/关闭的流实例
• 添加监控指标，跟踪活跃流数量、请求计数等关键数据

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