GCP Pubsub 低消息/秒的高延迟

Posted 2023-02-16

【中文标题】GCP Pubsub 低消息/秒的高延迟

【英文标题】：GCP Pubsub high latency on low message/sec

【发布时间】：2019-01-26 18:11:48

【问题描述】：

我正在使用 JAVA 客户端库从 AppEngine 灵活环境发布 Pubsub 消息，如下所示：

Publisher publisher = Publisher
                .newBuilder(ProjectTopicName.of(Utils.getApplicationId(), "test-topic"))
                .setBatchingSettings(
                        BatchingSettings.newBuilder()
                                .setIsEnabled(false)
                                .build())
                .build();

publisher.publish(PubsubMessage.newBuilder()
                .setData(ByteString.copyFromUtf8(message))
                .putAttributes("timestamp", String.valueOf(System.currentTimeMillis()))
                .build());

我正在订阅 Dataflow 中的主题并记录消息从 AppEngine flexible 到达 Dataflow 需要多长时间

pipeline
            .apply(PubsubIO.readMessagesWithAttributes().fromSubscription(Utils.buildPubsubSubscription(Constants.PROJECT_NAME, "test-topic")))
            .apply(ParDo.of(new DoFn<PubsubMessage, PubsubMessage>() 
                @ProcessElement
                public void processElement(ProcessContext c) 
                    long timestamp = System.currentTimeMillis() - Long.parseLong(c.element().getAttribute("timestamp"));
                    System.out.println("Time: " + timestamp);
                
            ));
    pipeline.run();

当我以每秒几条消息的速度发布消息时，日志显示消息到达 Dataflow 所需的时间在 100 毫秒到 1.5 秒之间。 但是当速率约为每秒 100 条消息时，时间总是在 100 毫秒 - 200 毫秒之间，这似乎完全足够了。 有人可以解释这种行为。似乎关闭发布者批处理不起作用。

【参考方案1】：

Pub/Sub 专为两种订阅情况下的高吞吐量消息而设计。

当有大量消息时，拉式订阅效果最好，当消息处理的吞吐量是您的优先事项时，您会使用这种订阅。特别注意，同步拉取不会一发布就处理消息，可以选择拉取和处理固定数量的消息（消息越多拉取越多）。更好的选择是使用异步拉取，它使用长时间运行的消息侦听器并一次确认一条消息 [1]。

另一方面，推送订阅使用慢启动算法：发送的消息数量随着每次成功交付而翻倍，直到达到其限制（更多消息，更多且更快的交付）。

---

[1]https://cloud.google.com/pubsub/docs/pull#asynchronous-pull

【讨论】：

据我所知，我不能将推送订阅与数据流一起使用。我也不能使用异步拉。因此，当我的主题没有足够的“真实”消息时，我应该创建空消息，以便更快地传递它们？