如何跟踪gRPC同步双向流式连接状态以可靠发送请求？

问题背景 #

使用gRPC同步双向流式通信，为省电设置流仅存活15秒。需要准确判断连接存活状态，确保能安全发送请求或缓存待处理请求；但现有拦截器+Ping方案不可靠，比如Wi-Fi切换蜂窝网络时IP变更，协议无错误抛出，同步连接无法等待，导致无法跟踪健康状态。

双向流定义：

ResponseStream<Update> connect(Stream<Request> request, {CallOptions? options})

当前初始化代码：

_channel = ClientChannel(
      baseUrl,
      port: ...,
      options: ChannelOptions(
        userAgent: userAgent,
        keepAlive: ClientKeepAliveOptions(
          pingInterval: Duration(seconds: 10),
          timeout: Duration(seconds: 3),
        ),
      ),
    );
    _customerStub = CustomerClient(
      channel,
      interceptors: [GRPCInterceptor()],
      options: CallOptionsBuilder()
          .setDeviceId(deviceId)
          .setClientToken(clientToken)
          .setAccessToken(_accessToken)
          .build(),
    );

_grpcGateway.customerStub
          .connect(_requestStreamController.stream)
          .listen(
        (update)...

解决方案 #

1. 利用gRPC内置回调捕获连接状态变化 #

同步流的listen回调包含onError和onDone，可以直接用来感知连接异常或正常结束：

• onError：连接断开、发送/接收失败时触发，此时标记连接为失效状态
• onDone：服务器主动关闭流（比如15秒存活到期）时触发，同样标记为失效

结合现有代码修改监听逻辑：

_grpcGateway.customerStub
    .connect(_requestStreamController.stream)
    .listen(
      (update) {
        // 收到响应说明连接正常，标记为活跃状态
        _currentStatus = ConnectionStatus.active;
        // 处理业务响应
      },
      onError: (error) {
        // 连接出现错误，标记为失效并触发重连
        _currentStatus = ConnectionStatus.failed;
        _scheduleReconnect();
      },
      onDone: () {
        // 流被主动关闭，标记为失效并触发重连
        _currentStatus = ConnectionStatus.failed;
        _scheduleReconnect();
      },
    );

2. 维护连接状态机+请求缓存 #

定义明确的连接状态枚举，在发送请求前检查状态，避免无效发送：

enum ConnectionStatus { idle, connecting, active, failed }

ConnectionStatus _currentStatus = ConnectionStatus.idle;
final List<Request> _pendingRequests = [];

void sendRequest(Request request) {
  switch (_currentStatus) {
    case ConnectionStatus.active:
      // 连接正常，直接发送
      _requestStreamController.add(request);
      break;
    case ConnectionStatus.idle:
    case ConnectionStatus.failed:
      // 连接未初始化或已失效，缓存请求并触发初始化/重连
      _pendingRequests.add(request);
      _initGrpcConnection();
      break;
    case ConnectionStatus.connecting:
      // 连接中，缓存请求等待连接完成
      _pendingRequests.add(request);
      break;
  }
}

在连接成功（收到第一个响应）时，批量发送缓存的请求：

// 在listen的onData回调中添加
if (_pendingRequests.isNotEmpty) {
  for (final req in _pendingRequests) {
    _requestStreamController.add(req);
  }
  _pendingRequests.clear();
}

3. 主动检测+指数退避重连兜底 #

针对网络切换无错误提示的场景，补充主动检测逻辑：

• 自定义Ping请求：定义空业务字段的Request（比如带isPing: true标识），在连接活跃时每隔12秒（小于15秒存活期）发送一次
• 服务器如果支持，返回对应的Ping响应；如果不支持，发送失败会触发onError，从而感知连接失效

重连时使用指数退避策略，避免频繁重连耗电：

int _retryCount = 0;

void _scheduleReconnect() {
  // 退避时间：1s, 2s, 4s...最大10s
  final delaySeconds = min(1 << _retryCount, 10);
  _retryCount++;
  Future.delayed(Duration(seconds: delaySeconds), () {
    _initGrpcConnection();
  });
}

// 重置重试次数：在连接成功时调用
void _resetRetryCount() {
  _retryCount = 0;
}

4. 监听系统网络事件 #

在移动端（如Flutter）借助网络状态监听库，当网络切换或恢复时，主动检查连接状态：

// 监听网络变化
Connectivity().onConnectivityChanged.listen((result) {
  if (result != ConnectivityResult.none && _currentStatus == ConnectionStatus.failed) {
    // 网络恢复，触发重连
    _initGrpcConnection();
  }
});

注意事项 #

• 确保服务器端配置匹配的KeepAlive参数，避免客户端Ping被拒绝
• 同步流的状态维护需保证线程安全（Flutter单线程环境下无需额外处理）
• 15秒存活期内，若没有业务请求，主动发送Ping保持连接，同时避免Ping过于频繁导致耗电

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