Flutter Cloud Pub Sub 如何创建可持续监听消息的StreamingPull？

一、StreamingPull实现方案 #

Google官方提供的googleapis Dart库基于REST接口实现，不支持基于gRPC双向流的StreamingPull能力，要实现StreamingPull需要直接调用Pub/Sub的gRPC接口，步骤如下：

• 第一步：添加依赖
  在pubspec.yaml中添加以下依赖：

dependencies:
  grpc: ^3.2.0
  protobuf: ^3.1.0
  googleapis_auth: ^1.4.1

• 第二步：生成Pub/Sub gRPC代码
  从Google官方API仓库下载Pub/Sub对应的proto定义文件，使用protoc命令配合Dart插件生成gRPC客户端代码，核心用到Subscriber服务的StreamingPull方法。
• 第三步：实现流监听逻辑
  核心示例代码如下：

// 初始化gRPC安全通道
final channel = ClientChannel(
  'pubsub.googleapis.com',
  port: 443,
  options: const ChannelOptions(credentials: ChannelCredentials.secure()),
);
final stub = SubscriberClient(channel);

// 构造请求流控制器
final requestStream = StreamController<StreamingPullRequest>();
// 发送初始拉取请求
requestStream.add(StreamingPullRequest(
  subscription: subscriptionName,
  streamAckDeadlineSeconds: 60,
));

// 监听服务端推送的消息流
stub.streamingPull(requestStream.stream).listen((response) {
  if (response.receivedMessages.isNotEmpty) {
    // 处理业务消息
    final ackIds = response.receivedMessages.map((msg) => msg.ackId).toList();
    // 提交确认回执
    requestStream.add(StreamingPullRequest(
      ackIds: ackIds,
    ));
  }
}, onError: (e) {
  // 自定义断连重连逻辑，可增加退避策略
  Future.delayed(const Duration(seconds: 5), () {
    // 重新初始化流连接
  });
});

注意需要自行实现权限token刷新、消息超时重发、异常重试等配套逻辑

二、其他可选实现方案 #

方案1：优化现有轮询逻辑 #

在你当前的单次拉取代码基础上封装循环轮询，实现成本极低，适合消息量不大的场景：

bool isPolling = true;
void startPolling() async {
  while (isPolling) {
    try {
      final pullResponse = await pubSubClient.projects.subscriptions.pull(
        PullRequest.fromJson({"maxMessages": 1000}),
        subscriptionName,
      );
      if (pullResponse.receivedMessages?.isNotEmpty == true) {
        final ids = pullResponse.receivedMessages!.map((e) => e.ackId).toList();
        await pubSubClient.projects.subscriptions.acknowledge(
          AcknowledgeRequest.fromJson({"ackIds": ids}),
          subscriptionName,
        );
        // 有消息时立即拉取下一批，无消息时延迟拉取降低请求量
        continue;
      }
    } catch (e) {
      // 出错时退避等待
      await Future.delayed(const Duration(seconds: 2));
    }
    await Future.delayed(const Duration(seconds: 1));
  }
}

方案2：Push订阅中转 #

在Google Cloud控制台将Pub/Sub订阅改为Push模式，配置消息推送到你自己的后端服务，后端再通过WebSocket、FCM等通道将消息推送到Flutter客户端，适合客户端量级大的场景，不需要客户端直接处理Pub/Sub协议。

方案3：云函数中转 #

配置Cloud Functions触发规则，当Pub/Sub收到消息时自动触发函数，将消息同步到Firebase Realtime Database或者Cloud Firestore，Flutter端直接监听数据库的实时变更即可，不需要自行维护长连接，实现成本最低。

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