如何确认gRPC双向流服务端发送的消息已成功送达所有客户端

gRPC 基于 HTTP/2 协议实现，默认只有传输层的 ACK 确认，仅能证明消息已经到达客户端的网络栈，无法证明上层业务逻辑已经接收并处理了消息，所以需要自行实现业务层面的回执机制来满足要求。

1. 调整 Proto 消息定义 #

首先在 .proto 文件中扩展消息结构，增加全局唯一消息ID和客户端回执字段：

syntax = "proto3";

package com.example.grpc;

service BidirectionalService {
  // 双向流接口
  rpc StreamConnect (stream ClientMessage) returns (stream ServerMessage) {}
}

// 服务端下发消息
message ServerMessage {
  int64 message_id = 1; // 全局唯一消息ID，用于匹配回执
  oneof payload {
    string common_msg = 2; // 示例业务消息，可扩展其他类型
  }
}

// 客户端上报消息
message ClientMessage {
  oneof payload {
    int64 ack_message_id = 1; // 确认收到的消息ID，用于回执
    // 可扩展其他客户端上报的业务字段
  }
}

2. 服务端实现会话管理 #

服务端需要维护所有活跃的双向流连接，以及每条消息的回执状态，示例代码如下：

import io.grpc.stub.StreamObserver;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

// 全局会话管理器，线程安全
public class SessionManager {
  // 存储所有活跃客户端的流实例，key为客户端唯一标识（如clientId）
  private final ConcurrentHashMap<String, StreamObserver<ServerMessage>> activeSessions = new ConcurrentHashMap<>();
  // 存储消息ID对应的已回执客户端集合
  private final ConcurrentHashMap<Long, Set<String>> ackedClients = new ConcurrentHashMap<>();
  // 全局消息ID生成器
  private final AtomicLong messageIdGenerator = new AtomicLong(0);

  // 注册新的客户端连接
  public void registerSession(String clientId, StreamObserver<ServerMessage> observer) {
    activeSessions.put(clientId, observer);
  }

  // 移除断开的客户端连接
  public void removeSession(String clientId) {
    activeSessions.remove(clientId);
  }
}

3. 广播消息+等待全量回执逻辑 #

发送消息时先记录当前所有在线客户端，广播后阻塞等待所有回执，可通过CountDownLatch实现阻塞逻辑：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public void broadcastAndWaitForAllAck(String messageContent) throws InterruptedException {
  long messageId = messageIdGenerator.incrementAndGet();
  Set<String> currentOnlineClients = new HashSet<>(activeSessions.keySet());
  int clientCount = currentOnlineClients.size();
  
  // 无在线客户端直接执行后续逻辑
  if (clientCount == 0) {
    runNextLogic();
    return;
  }

  // 初始化当前消息的回执存储
  ackedClients.put(messageId, ConcurrentHashMap.newKeySet());
  CountDownLatch ackLatch = new CountDownLatch(clientCount);

  // 构造广播消息
  ServerMessage broadcastMsg = ServerMessage.newBuilder()
    .setMessageId(messageId)
    .setCommonMsg(messageContent)
    .build();

  // 向所有在线客户端发送消息
  for (Map.Entry<String, StreamObserver<ServerMessage>> entry : activeSessions.entrySet()) {
    String clientId = entry.getKey();
    StreamObserver<ServerMessage> stream = entry.getValue();
    try {
      stream.onNext(broadcastMsg);
    } catch (Exception e) {
      // 发送失败说明客户端已断连，直接计数
      ackLatch.countDown();
      ackedClients.get(messageId).add(clientId);
    }
  }

  // 注册回执回调：需在客户端流的onNext方法中收到ACK时调用
  Runnable ackCallback = (clientId, ackMsgId) -> {
    if (ackMsgId == messageId && ackedClients.get(messageId).add(clientId)) {
      ackLatch.countDown();
    }
  };
  // 此处将回调注册到全局回执处理器即可，收到对应messageId的ACK时触发

  // 等待所有回执，设置30秒超时避免永久阻塞
  boolean allAcked = ackLatch.await(30, TimeUnit.SECONDS);
  // 清理临时存储
  ackedClients.remove(messageId);

  if (allAcked) {
    // 所有客户端已确认，执行后续逻辑
    runNextLogic();
  } else {
    // 超时未收到全量回执，可根据业务需要做重试、告警等处理
    handleAckTimeout();
  }
}

4. 客户端实现回执逻辑 #

客户端收到服务端消息后，立即返回对应消息ID的ACK即可：

// 客户端侧双向流实现示例
StreamObserver<ServerMessage> serverStreamObserver = new StreamObserver<ServerMessage>() {
  @Override
  public void onNext(ServerMessage serverMsg) {
    // 处理业务消息
    System.out.println("收到服务端消息: " + serverMsg.getCommonMsg());
    // 回传ACK
    ClientMessage ackMsg = ClientMessage.newBuilder()
      .setAckMessageId(serverMsg.getMessageId())
      .build();
    clientSendStream.onNext(ackMsg);
  }

  @Override
  public void onError(Throwable t) {
    // 异常处理逻辑
  }

  @Override
  public void onCompleted() {
    // 流关闭处理逻辑
  }
};

注意事项 #

• 必须设置超时机制，避免单个客户端断连未上报导致服务端永久阻塞
• 如果业务允许客户端离线，可自行扩展离线消息存储、客户端上线后补发的逻辑
• 消息ID要保证全局唯一，避免ACK匹配错误
• 会话管理器需要做好线程安全处理，gRPC的流回调默认在不同线程执行，避免并发问题

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