问题标签 [google-cloud-pubsub]

我有一个需要非常低延迟（实时游戏）的应用程序。目前在我的解决方案中，消息从客户端前端服务器路由到目标服务器所需的时间不到 2 毫秒。

有人知道 Google Cloud Pub/Sub 将消息从一台服务器路由到另一台服务器需要多长时间吗？谢谢！

我的测试管道非常简单。它尝试从通过控制台创建的主题中读取。

当我尝试执行此管道时，出现以下错误：

工作流失败。原因：（de5f777e2e08c1d9）：步骤 setup_resource_additionaltopic.subscription-375367840492394866711：设置资源 additionaltopic.subscription-3753678404923948667 失败

Dataflow 控制台还报告了一个内部错误：

我在文档中找不到任何内容，并且我尝试解决此问题的尝试和错误尝试都没有成功。

解决方案

要运行 Dataflow 作业，项目必须启用以下 Google Cloud Platform API：

• 谷歌云数据流 API
• 计算引擎 API（谷歌计算引擎）
• 谷歌云记录 API
• 谷歌云存储
• 谷歌云存储 JSON API
• BigQuery API
• 谷歌云发布/订阅
• 谷歌云数据存储 API

您可以使用 Google Cloud Platform Console一次性启用所有必需的 API。

从 google-dataproc 集群使用/安装 python google cloud api（例如 pub-sub）的正确方法是什么？例如，如果我在集群上使用 zeppelin/pyspark 并且我想使用 pub-sub api，我应该如何准备它？

我不清楚在默认集群配置期间安装了什么和未安装什么，以及我是否应该/如何尝试为谷歌云 api 安装 python 库。

我还意识到可能需要设置范围/身份验证。需要明确的是，我可以在本地使用 api，但我不确定从集群访问 api 的最干净的方法是什么，我不想执行任何不必要的步骤。

我正在尝试针对本地模拟 pubsub 服务器使用Go pubsub 库。我发现“旧样式”（已弃用）函数（例如and ）可以正常工作，但“新样式”API（例如and ）无法按预期工作。CreateSubPullWaitIteratorsSubscriptionHandles

我编写了两个不同的单元测试，它们都测试相同的动作序列，一个使用“新风格”API，一个使用“旧风格”API。

顺序是：

• 创建订阅
• 无法提取任何消息（因为没有可用的消息）
• 发布消息
• 拉该消息，但不要确认它
• 最后再拉一次，这需要 10 秒，因为消息 ACK 超时必须先过期

https://gist.github.com/ianrose14/db6ecd9ccb6c84c8b36bf49d93b11bfb

使用旧式 API 的测试正如我所期望的那样工作：

而使用新型 API 的测试工作不可靠。有时事情会按预期工作：

但有时我发现它iter.Stop()并没有像它应该的那样迅速返回（并注意第二个 iter.Next 比它应该长得多）：

而其他时候我发现发布消息后的第一次拉取时间太长（它应该接近即时）：

有任何想法吗？是否有任何使用新型 API 的工作示例？不幸的是，这里的 Go 入门项目使用了旧的、已弃用的 API。

有没有办法确定应用引擎收到的消息来自 Google PubSub 服务？目前，PubSub 服务在 appengine app.yaml 中配置为“login: admin”的 URL 上获得 302。所以它一直在重试。

我本来希望它的行为类似于 Appengine 中的任务，并自动对“登录：管理员”URL 进行身份验证。

我正在尝试使用 Google Cloud Dataflow 将 Google PubSub 消息写入 Google Cloud Storage。我知道 TextIO/AvroIO 不支持流式传输管道。ParDo/DoFn但是，我在 [1] 中读到，可以从作者的评论中的流式管道中写入 GCS 。我尽可能地按照他们的文章构建了一个管道。

我的目标是这种行为：

• 在与消息发布时间相对应的路径下，将最多 100 条消息批量写入 GCS 中的对象（每个窗口窗格一个）dataflow-requests/[isodate-time]/[paneIndex]。

我得到不同的结果：

• 每个小时窗口中只有一个窗格。因此，我每小时只得到一个文件（它实际上是 GCS 中的一个对象路径）。将 MAX_EVENTS_IN_FILE 减少到 10 没有任何区别，仍然只有一个窗格/文件。
• 每个 GCS 对象中只有一条消息被写出
• 在写入 GCS 时，管道偶尔会引发 CRC 错误。

如何解决这些问题并获得我期望的行为？

示例日志输出：

这是我的代码：

[1] https://labs.spotify.com/2016/03/10/spotifys-event-delivery-the-road-to-the-cloud-part-iii/

感谢 Sam McVeety 的解决方案。这是任何阅读的人的更正代码：

我们正在从 AMQP 迁移到 Google 的 Pubsub。

文档建议拉可能是我们的最佳选择，因为我们正在使用计算引擎并且无法打开我们的工作人员以通过推送服务接收。

它还表示，根据使用情况，拉动可能会产生额外的成本：

如果使用轮询，如果您频繁打开连接并立即关闭它们，可能会导致高网络使用率。

我们在 go 中创建了一个测试订阅者，它在循环中运行，如下所示：

我的问题是这是否是正确/推荐的方式来获取消息。

我们全天每秒收到大约 100 条消息。我不确定在无限循环中运行它是否会让我们破产，并且找不到任何其他像样的 go 示例。

GCE 实例组可以根据 Google Cloud PubSub 队列计数或其他异步任务队列（例如 PSQ）进行扩展/缩减吗？

在 Google Pub/Sub 中，是否可以使用在 Compute Engine VM 实例上运行的 HTTPS 服务器作为推送订阅的端点？我有点难过，因为在文档中它说服务器必须可以通过 DNS 名称访问。

任何建议表示赞赏 - 干杯！

我是 Google Compute/Google App Engine 平台的新手。我目前正在将使用 celery 执行异步任务的 python 烧瓶应用程序迁移到 Google Compute/Google App Engine 平台。但是在文档中，我应该使用 Google Pub/Sub 而不是 celery。在我的应用程序中，每当我运行异步任务时，我都会使用与http://blog.miguelgrinberg.com/post/using-celery-with-flask相同的原理来监视作业状态的页面。我已经检查了 google pub/sub 的文档，但是我不知道如何使用 google pub/sub 来实现它。任何人都可以帮助或指出我在谷歌发布/订阅中实现相同的正确方向。