构建一对多 Pub/Sub 系统

本教程介绍了如何设置一组应用，这些应用通过 Pub/Sub（而不是同步 RPC）发送消息来进行通信。通过分离应用，消息传递：

• 使应用更稳健
• 可以简化开发

例如，调用方（发布者）不需要接收方（订阅者）随时可用。它只是向 Pub/Sub 发送消息。发布者也不需要知道需要接收消息的订阅者应用具体是哪个以及应用的数量是多少。因此，可以依赖该服务在消息可用时将消息传送给一个或多个订阅者应用。

系统概览

在本教程中，您将启动一个发布者应用，该应用使用一对多通信向两个订阅者发送“Hello, World!”消息，如下图所示：

这两个订阅者应用使用相同的代码，但您可以在不同的时间启动它们。此过程演示了 Pub/Sub 启用异步通信的方式。如需构建此系统，请完成以下步骤：

1. 创建应用用于身份验证的 IAM 服务帐号。
2. 设置 IAM 权限
3. 创建 Pub/Sub 主题和订阅。
4. 启动三个独立应用：一个发布者应用和两个订阅者应用

准备工作

1. 登录您的 Google Cloud 账号。如果您是 Google Cloud 新手，请创建一个账号来评估我们的产品在实际场景中的表现。新客户还可获享 $300 赠金，用于运行、测试和部署工作负载。
2. 安装 Google Cloud CLI。

3. 如需初始化 gcloud CLI，请运行以下命令：

   gcloud init

4. 创建或选择 Google Cloud 项目。

   注意：如果您不打算保留在此过程中创建的资源，请创建新的项目，而不要选择现有的项目。完成本教程介绍的步骤后，您可以删除所创建的项目，并移除与该项目关联的所有资源。

   • 创建 Google Cloud 项目：

     gcloud projects create PROJECT_ID

     将 PROJECT_ID 替换为您要创建的 Google Cloud 项目的名称。

   • 选择您创建的 Google Cloud 项目：

     gcloud config set project PROJECT_ID

     将 PROJECT_ID 替换为您的 Google Cloud 项目 名称。

5. 确保您的 Google Cloud 项目已启用结算功能。

6. Enable the Pub/Sub API:

   gcloud services enable pubsub.googleapis.com

7. 为您的 Google 账号创建本地身份验证凭据：

   gcloud auth application-default login

8. 向您的 Google 账号授予角色。对以下每个 IAM 角色运行以下命令一次： roles/pubsub.publisher, roles/pubsub.subscriber

   gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

   • 将 PROJECT_ID 替换为您的项目 ID。
   • 将 EMAIL_ADDRESS 替换为您的电子邮件地址。
   • 将 ROLE 替换为每个角色。
9. 安装 Google Cloud CLI。

10. 如需初始化 gcloud CLI，请运行以下命令：

    gcloud init

11. 创建或选择 Google Cloud 项目。

    注意：如果您不打算保留在此过程中创建的资源，请创建新的项目，而不要选择现有的项目。完成本教程介绍的步骤后，您可以删除所创建的项目，并移除与该项目关联的所有资源。

    • 创建 Google Cloud 项目：

      gcloud projects create PROJECT_ID

      将 PROJECT_ID 替换为您要创建的 Google Cloud 项目的名称。

    • 选择您创建的 Google Cloud 项目：

      gcloud config set project PROJECT_ID

      将 PROJECT_ID 替换为您的 Google Cloud 项目 名称。

12. 确保您的 Google Cloud 项目已启用结算功能。

13. Enable the Pub/Sub API:

    gcloud services enable pubsub.googleapis.com

14. 为您的 Google 账号创建本地身份验证凭据：

    gcloud auth application-default login

15. 向您的 Google 账号授予角色。对以下每个 IAM 角色运行以下命令一次： roles/pubsub.publisher, roles/pubsub.subscriber

    gcloud projects add-iam-policy-binding PROJECT_ID --member="user:EMAIL_ADDRESS" --role=ROLE

    • 将 PROJECT_ID 替换为您的项目 ID。
    • 将 EMAIL_ADDRESS 替换为您的电子邮件地址。
    • 将 ROLE 替换为每个角色。

安装 Python

本教程使用 Pub/Sub 客户端库，它需要 Python 3.7 或更高版本。完成安装 Python 的说明。

设置您的 Pub/Sub 项目

如需管理发布和订阅应用之间的消息流，您需要创建一个主题和两个不同的订阅。

创建 Pub/Sub 主题

创建 ID 为 hello_topic 的主题：

gcloud pubsub topics create hello_topic

创建 Pub/Sub 订阅

创建两个订阅并将其附加到您的主题。

这些订阅是 StreamingPull 订阅，也是一种pull订阅。

订阅 1

创建 ID 为 sub_one 的订阅并将其附加到 hello_topic。

gcloud pubsub subscriptions create sub_one --topic=hello_topic

订阅 2

创建 ID 为 sub_two 的订阅并将其附加到 hello_topic。

gcloud pubsub subscriptions create sub_two --topic=hello_topic

构建一对多系统

下载发布商和订阅者代码

1. 下载本教程所需的 Pub/Sub Python 文件。

    git clone https://github.com/googleapis/python-pubsub.git
   

2. 关闭所有打开的终端，然后再继续操作。

设置三个终端

1. 为每个教程应用（一个发布者应用和两个订阅者应用）启动一个终端。为方便起见，本教程将调用以下终端：

   • 发布者终端
   • sub_onesub_one 终端
   • sub_twosub_two 终端
   注意 ：确保所有三个终端都位于与您在上一步中下载 Pub/Sub Python 文件的位置位于同一目录中。

2. 在发布者终端中，创建并激活名为 pyenv-qs 的 Python 虚拟环境。

   Bash

   python -m venv pyenv-qs
   source pyenv-qs/bin/activate
   

   PowerShell

   py -m venv pyenv-qs
   .\pyenv-qs\Scripts\activate
   

   在 sub_onesub_one 和 sub_one 终端中，运行以下命令：

   Bash

   source pyenv-qs/bin/activate
   

   PowerShell

   .\pyenv-qs\Scripts\activate
   

   运行 activate 命令后，命令提示符会包含以下值 (pyenv-qs) $。

3. 在发布者终端中，使用 pip 安装 Pub/Sub Python 客户端库：

   python -m pip install --upgrade google-cloud-pubsub

4. 在这三个终端中，使用当前项目 ID 设置环境变量。以下 gcloud 命令确定您选择的项目 ID 并将其设置为变量：

   Bash

   export PROJECT=`gcloud config get-value project`
   

   PowerShell

   $env:PROJECT=$(gcloud config get-value project)
   

5. 在这三个终端中，切换到包含示例代码的项目路径。

   cd python-pubsub/samples/snippets/quickstart/
   

启动应用并观察消息流

启动订阅者 1 应用

在 sub_onesub_one 终端中，启动订阅者 1：

Bash

python sub.py $PROJECT sub_one

PowerShell

py sub.py $env:PROJECT sub_one

启动后，此应用会打开与服务器之间的双向流式传输连接。Pub/Sub 通过数据流传送消息。

启动发布者应用

在发布者终端中，启动发布者应用：

Bash

python pub.py $PROJECT hello_topic

PowerShell

py pub.py $env:PROJECT hello_topic

发布者应用启动后，Pub/Sub 系统将执行以下操作：

• 发布者应用将“Hello, World!”消息发送给 Pub/Sub，此时该应用还不知道现在是否有任何订阅。该服务器还分配一个消息 ID。

• 订阅者 1 应用接收“Hello World”消息，输出该消息，并向 Pub/Sub 发送确认。

• 发布者应用输出确认。确认告知 Pub/Sub 消息已成功处理，无需重新发送给此订阅者或任何其他 sub_onesub_one 订阅者。

Pub/Sub 移除 sub_one sub_one中的消息。

启动订阅者 2 应用

在 sub_twosub_two 终端中，启动订阅者 2：

Bash

python sub.py $PROJECT sub_two

PowerShell

py sub.py $env:PROJECT sub_two

该订阅者接收传递给 sub_twosub_two 订阅的消息。 订阅者 2 重复使用 sub.py 脚本。不同之处在于，订阅者 2 在发布者将消息发送到主题和订阅之后才会启动。如果发布者直接调用订阅者 2，则发布应用要么等到订阅者 2 启动，要么就会超时。Pub/Sub 通过有效地为订阅者 2 保存该消息来管理此过程。

现在，您可以使用 Pub/Sub 进行开发了！

结果怎么样？

成功了！

太棒了！ 您对本快速入门有何看法？我们可以在哪些方面做出改进？ 欢迎提供您的意见！

遇到了问题！

我们对此深表遗憾。 请告诉我们您遇到了什么问题。我们会予以修正。

Pub/Sub 支持页面提供了其他资源和链接。

清理

1. 停止所有正在运行的应用。
2. 从本地环境中删除示例代码目录。

3. 删除主题。

   gcloud pubsub topics delete hello_topic
   

4. 删除订阅。

   gcloud pubsub subscriptions delete sub_one
   

   gcloud pubsub subscriptions delete sub_two
   

5. 在 Google Cloud 控制台的 IAM 和管理部分关闭教程项目。

6. 可选：撤消您创建的身份验证凭据，并删除本地凭据文件。

   gcloud auth application-default revoke

7. 可选：从 gcloud CLI 撤消凭据。

   gcloud auth revoke

后续步骤

以下是您可以尝试的一些事项：

• 检查教程的 pub.py 和 sub.py 代码，并在 GitHub 上浏览其他 Pub/Sub 示例。作为练习，请创建一个每秒发布一次本地时间的 pub.py 版本。

• 了解如何批量处理消息。

• 使用推送订阅，接收触发 App Engine 端点或 Cloud Functions 函数的消息。

• 使用重放检索以前确认的消息。 默认情况下，Pub/Sub 会从订阅中移除已确认的消息。例如，在本教程中，您将无法重新运行 sub.py 来再次接收“Hello, World!”消息。借助重放功能，您可以设置订阅，以便在确认消息后接收消息。

• 开始使用其他语言的客户端库。