从 App Engine 任务队列拉取任务迁移到 Cloud Pub/Sub（模块 19）

1. 概览

无服务器迁移站系列 Codelab（自定进度的动手教程）和相关视频旨在帮助 Google Cloud 无服务器开发者通过一次或多次迁移（主要是从旧版服务迁移）来指导他们的应用现代化。这样做可让您的应用更易于移植，并为您提供更多选择和灵活性，使您能够与更多 Cloud 产品集成并访问这些产品，还能更轻松地升级到新的语言版本。虽然最初侧重于最早的 Cloud 用户（主要是 App Engine [标准环境] 开发者），但本系列文章的范围足够广，可涵盖其他无服务器平台（例如 Cloud Functions 和 Cloud Run），或者其他平台（如果适用）。

此 Codelab 的目的是向 Python 2 App Engine 开发者展示如何从 App Engine 任务队列拉取任务迁移到 Cloud Pub/Sub。此外，对于 Datastore 访问（主要在模块 2 中介绍），还有从 App Engine NDB 到 Cloud NDB 的隐式迁移，以及升级到 Python 3。

在模块 18 中，您将学习如何在应用中添加对拉取任务的使用。在本模块中，您将使用完成的模块 18 应用，并将该使用迁移到 Cloud Pub/Sub。如果使用 Task Queues 处理推送任务，则应改用 Cloud Tasks，并参阅模块 7-9。

在接下来的实验中

将 App Engine Task Queue（拉取任务）的使用替换为 Cloud Pub/Sub
将 App Engine NDB 的使用替换为 Cloud NDB（另请参阅模块 2）
将应用移植到 Python 3

所需条件

具有有效的 GCP 结算账号的 Google Cloud Platform 项目
基本 Python 技能
常用 Linux 命令的实践知识
具备开发和部署 App Engine 应用的基础知识
有效的模块 18 App Engine 示例应用

调查问卷

您打算如何使用本教程？

仅通读

阅读并完成练习

您如何评价使用 Python 的体验？

新手水平

中等水平

熟练水平

您如何评价自己在使用 Google Cloud 服务方面的经验水平？

新手水平

中等水平

熟练水平

2. 背景

App Engine Task Queue 同时支持推送任务和拉取任务。为了提高应用的可移植性，Google Cloud 建议从任务队列等旧版捆绑服务迁移到其他 Cloud 独立服务或第三方等效服务。

任务队列推送任务用户应迁移到 Cloud Tasks。
任务队列拉取任务的用户应迁移到 Cloud Pub/Sub。

模块 7-9 介绍了推送任务迁移，而模块 18-19 则重点介绍了拉取任务迁移。虽然 Cloud Tasks 与任务队列推送任务的匹配度更高，但 Pub/Sub 与任务队列拉取任务的相似度并不高。

Pub/Sub 提供的功能比任务队列提供的拉取功能更多。例如，Pub/Sub 也具有推送功能，但 Cloud Tasks 更像 Task Queue 推送任务，因此任何迁移模块都不涵盖 Pub/Sub 推送。此模块 19 Codelab 演示了如何将排队机制从任务队列拉取队列切换到 Pub/Sub，以及如何从 App Engine NDB 迁移到 Cloud NDB 以进行数据存储区访问，重复模块 2 的迁移。

虽然模块 18 中的代码“宣传”为 Python 2 示例应用，但源代码本身与 Python 2 和 3 兼容，即使在模块 19 中迁移到 Cloud Pub/Sub（和 Cloud NDB）后，仍然如此。

本教程包含以下步骤：

设置/准备工作
更新配置
修改应用代码

3. 设置/准备工作

本节介绍如何执行以下操作：

设置 Cloud 项目
获取基准示例应用
（重新）部署并验证基准应用
启用新的 Google Cloud 服务/API

这些步骤可确保您从可正常运行的代码开始，并确保代码已准备好迁移到云服务。

1. 设置项目

如果您已完成模块 18 Codelab，请重复使用同一项目（和代码）。或者，您可以创建一个全新的项目或重复使用另一个现有项目。确保项目具有有效的结算账号，并且已启用 App Engine 应用。找到您的项目 ID，因为您需要在本 Codelab 中随时使用它，每当遇到 PROJECT_ID 变量时，都要使用该 ID。

2. 获取基准示例应用

前提条件之一是拥有一个正常运行的模块 18 App Engine 应用，因此请完成其 Codelab（推荐；上面的链接），或从代码库中复制模块 18 代码。无论您是使用自己的代码还是我们的代码，我们都能在这里开始（“START”）。本 Codelab 将引导您完成迁移，最后获得类似于模块 19 代码库文件夹（“FINISH”）中的代码。

START：模块 18 文件夹 (Python 2)
完成：模块 19 文件夹 (Python 2 和 3)
整个代码库（用于克隆或下载 ZIP 文件）

无论您使用哪个模块 18 应用，该文件夹都应如下所示，可能还会包含 lib 文件夹：

$ ls
README.md               appengine_config.py     queue.yaml              templates
app.yaml                main.py                 requirements.txt

3. （重新）部署并验证基准应用

执行以下步骤来部署模块 18 应用：

删除 lib 文件夹（如果有），然后运行 pip install -t lib -r requirements.txt 以重新填充 lib。如果您的开发机器上同时安装了 Python 2 和 3，您可能需要改用 pip2。
确保您已安装并初始化 gcloud 命令行工具，并查看其用法。
（可选）使用 gcloud config set project PROJECT_ID 设置您的云项目，这样您就不必在每次发出 gcloud 命令时都输入 PROJECT_ID。
使用 gcloud app deploy 部署示例应用
确认应用按预期正常运行。如果您已完成模块 18 的 Codelab，应用会显示热门访问者以及最近的访问（如下所示）。否则，可能没有要显示的访问者数量。

在迁移模块 18 示例应用之前，您必须先启用修改后的应用将使用的 Cloud 服务。

4. 启用新的 Google Cloud 服务/API

旧版应用使用 App Engine 捆绑服务，无需进行额外设置，但独立 Cloud 服务需要进行额外设置，并且更新后的应用将同时使用 Cloud Pub/Sub 和 Cloud Datastore（通过 Cloud NDB 客户端库）。App Engine 和这两个 Cloud API 都有“永久免费”层级配额，只要您不超过这些限制，完成本教程就不会产生费用。您可以根据自己的偏好，通过 Cloud 控制台或命令行启用 Cloud API。

通过 Cloud 控制台

在 Cloud 控制台中，前往 API 管理器的“库”页面（确保选择正确的项目），然后使用页面中间的搜索栏搜索 Cloud Datastore 和 Cloud Pub/Sub API：

为每个 API 单独点击启用按钮 - 系统可能会提示您提供结算信息。例如，以下是 Cloud Pub/Sub API 库页面：

从命令行

虽然通过控制台启用 API 在视觉上更直观，但有些人更喜欢使用命令行。发出 gcloud services enable pubsub.googleapis.com datastore.googleapis.com 命令以同时启用这两个 API：

$ gcloud services enable pubsub.googleapis.com datastore.googleapis.com
Operation "operations/acat.p2-aaa-bbb-ccc-ddd-eee-ffffff" finished successfully.

系统可能会提示您输入结算信息。如果您想启用其他 Cloud API，并想知道它们的 URI，可以在每个 API 的库页面底部找到这些信息。例如，请注意上文 Pub/Sub 页面底部的“服务名称”pubsub.googleapis.com。

完成这些步骤后，您的项目将能够访问这些 API。现在，您可以更新应用以使用这些 API 了。

4. 创建 Pub/Sub 资源

回顾一下模块 18 中任务队列工作流的顺序：

模块 18 使用 queue.yaml 文件创建了一个名为 pullq 的拉取队列。
应用会将任务添加到拉取队列，以跟踪访问者。
任务最终由工作器处理，租用时间有限（一小时）。
执行任务以统计近期访问者数量。
任务完成后，系统会将其从队列中删除。

您将使用 Pub/Sub 复制类似的工作流。下一部分将介绍基本的 Pub/Sub 术语，并提供三种创建必要 Pub/Sub 资源的不同方法。

App Engine 任务队列（拉取）与 Cloud Pub/Sub 术语对比

改用 Pub/Sub 需要对词汇稍作调整。下面列出了主要类别以及这两个产品中的相关术语。另请查看迁移指南，其中也包含类似比较。

排队数据结构：使用任务队列时，数据会进入拉取队列；使用 Pub/Sub 时，数据会进入主题。
排队的数据单位： Task Queue 中的拉取任务在 Pub/Sub 中称为消息。
数据处理器：使用 Task Queue 时，工作器可访问拉取任务；使用 Pub/Sub 时，您需要订阅/订阅者来接收消息
数据提取： 租用拉取任务与通过订阅从主题拉取消息相同。
清理/完成： 完成任务后，从拉取队列中删除 Task Queue 任务，这类似于确认 Pub/Sub 消息

虽然排队产品有所变化，但工作流程仍然相对相似：

该应用使用的是名为 pullq 的主题，而不是拉取队列。
应用不会将任务添加到拉取队列，而是将消息发送到主题 (pullq)。
与工作器从拉取队列中租用任务不同，名为 worker 的订阅者从 pullq 主题拉取消息。
应用处理消息载荷，并递增数据存储区中的访问者数量。
应用不会从拉取队列中删除任务，而是确认已处理的消息。

使用任务队列时，设置过程包括创建拉取队列。使用 Pub/Sub 时，设置需要同时创建主题和订阅。在模块 18 中，我们处理了应用执行之外的 queue.yaml；现在，我们必须对 Pub/Sub 执行相同的操作。

您可以通过以下三种方式创建主题和订阅：

通过 Cloud 控制台
从命令行，或
通过代码（简短的 Python 脚本）

选择以下任一选项，然后按照相应说明创建 Pub/Sub 资源。

通过 Cloud 控制台

如需通过 Cloud Console 创建主题，请按以下步骤操作：

前往 Cloud 控制台中的 Pub/Sub 主题页面。
点击顶部的创建主题；系统会打开一个新的对话框窗口（见下图）
在主题 ID 字段中，输入 pullq。
取消选中所有已勾选的选项，然后选择 Google 管理的加密密钥。
点击“创建”主题按钮。

主题创建对话框如下所示：

现在您已创建主题，接下来必须为该主题创建订阅：

前往 Cloud 控制台中的 Pub/Sub 订阅页面。
点击顶部的创建订阅（见下图）。
在订阅 ID 字段中输入 worker。
从选择 Cloud Pub/Sub 主题下拉菜单中选择 pullq，并记下其“完全限定的路径名”，例如 projects/PROJECT_ID/topics/pullq
对于传送类型，选择拉取。
将所有其他选项保留原样，然后点击创建按钮。

订阅创建界面如下所示：

您还可以通过主题页面创建订阅，此“快捷方式”可能有助于您将主题与订阅相关联。如需详细了解如何创建订阅，请参阅文档。

从命令行

Pub/Sub 用户可以使用命令 gcloud pubsub topics create TOPIC_ID 和 gcloud pubsub subscriptions create SUBSCRIPTION_ID --topic=TOPIC_ID 分别创建主题和订阅。如果以 TOPIC_ID 为 pullq 且 SUBSCRIPTION_ID 为 worker 执行这些命令，则项目 PROJECT_ID 的输出如下：

$ gcloud pubsub topics create pullq
Created topic [projects/PROJECT_ID/topics/pullq].

$ gcloud pubsub subscriptions create worker --topic=pullq
Created subscription [projects/PROJECT_ID/subscriptions/worker].

另请参阅快速入门文档中的此页面。如果需要定期创建主题和订阅，使用命令行可能会简化工作流程，并且此类命令可用于 shell 脚本。

通过代码（简短的 Python 脚本）

另一种自动创建主题和订阅的方法是在源代码中使用 Pub/Sub API。以下是模块 19 代码库文件夹中 maker.py 脚本的代码。

from __future__ import print_function
import google.auth
from google.api_core import exceptions
from google.cloud import pubsub

_, PROJECT_ID = google.auth.default()
TOPIC = 'pullq'
SBSCR = 'worker'
ppc_client = pubsub.PublisherClient()
psc_client = pubsub.SubscriberClient()
TOP_PATH = ppc_client.topic_path(PROJECT_ID, TOPIC)
SUB_PATH = psc_client.subscription_path(PROJECT_ID, SBSCR)

def make_top():
    try:
        top = ppc_client.create_topic(name=TOP_PATH)
        print('Created topic %r (%s)' % (TOPIC, top.name))
    except exceptions.AlreadyExists:
        print('Topic %r already exists at %r' % (TOPIC, TOP_PATH))

def make_sub():
    try:
        sub = psc_client.create_subscription(name=SUB_PATH, topic=TOP_PATH)
        print('Subscription created %r (%s)' % (SBSCR, sub.name))
    except exceptions.AlreadyExists:
        print('Subscription %r already exists at %r' % (SBSCR, SUB_PATH))
    try:
        psc_client.close()
    except AttributeError:  # special Py2 handler for grpcio<1.12.0
        pass

make_top()
make_sub()

执行此脚本会生成预期输出（前提是没有错误）：

$ python3 maker.py
Created topic 'pullq' (projects/PROJECT_ID/topics/pullq)
Subscription created 'worker' (projects/PROJECT_ID/subscriptions/worker)

调用 API 来创建已存在的资源会导致客户端库抛出 google.api_core.exceptions.AlreadyExists 异常，该异常会被脚本妥善处理：

$ python3 maker.py
Topic 'pullq' already exists at 'projects/PROJECT_ID/topics/pullq'
Subscription 'worker' already exists at 'projects/PROJECT_ID/subscriptions/worker'

如果您刚接触 Pub/Sub，请参阅 Pub/Sub 架构白皮书，以获取更多见解。

5. 更新配置

配置更新包括更改各种配置文件，以及在 Cloud Pub/Sub 生态系统中创建相当于 App Engine 拉取队列的队列。

删除 queue.yaml

我们将完全弃用任务队列，因此删除了 queue.yaml，因为 Pub/Sub 不使用此文件。您将创建 Pub/Sub 主题（和订阅），而不是创建拉取队列。

requirements.txt

将 google-cloud-ndb 和 google-cloud-pubsub 都附加到 requirements.txt，以便加入模块 18 中的 flask。更新后的模块 19 requirements.txt 现在应如下所示：

flask
google-cloud-ndb
google-cloud-pubsub

此 requirements.txt 文件不包含任何版本号，这意味着系统会选择最新版本。如果出现任何不兼容情况，请按照标准做法使用版本号来锁定应用的有效版本。

app.yaml

对 app.yaml 的更改因您是继续使用 Python 2 还是升级到 Python 3 而异。

Python 2

上述对 requirements.txt 的更新添加了对 Google Cloud 客户端库的使用。这些功能需要 App Engine 提供额外的支持，即两个内置库：setuptools 和 grpcio。使用内置库需要在 app.yaml 中添加 libraries 部分，并指定库版本号，或者指定“latest”以使用 App Engine 服务器上提供的最新版本。模块 18 app.yaml 还没有以下部分：

之前：

runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

向 app.yaml 添加 libraries 部分，并添加 setuptools 和 grpcio 的条目，选择它们的最新版本。此外，还添加了一个 Python 3 的占位符 runtime 条目，并将其与当前的 3.x 版本（例如撰写本文时的 3.10）一起注释掉。完成这些更改后，app.yaml 现在如下所示：

升级后：

#runtime: python310
runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

libraries:
- name: setuptools
  version: latest
- name: grpcio
  version: latest

Python 3

对于 Python 3 用户和 app.yaml，一切都与移除有关。在本部分中，您将删除 handlers 部分、threadsafe 和 api_version 指令，并且不会创建 libraries 部分。

第二代运行时不提供内置的第三方库，因此 app.yaml 中不需要 libraries 部分。此外，不再需要复制（有时称为供应商化或自打包）非内置的第三方软件包。您只需在 requirements.txt 中列出应用使用的第三方库。

app.yaml 中的 handlers 部分用于指定应用（脚本）和静态文件处理程序。由于 Python 3 运行时要求 Web 框架执行自己的路由，因此所有脚本处理程序都必须更改为 auto。如果您的应用（如模块 18 中的应用）不提供静态文件，则所有路由都将为 auto，从而变得无关紧要。因此，也不需要 handlers 部分，请将其删除。

最后，Python 3 中不使用 threadsafe 和 api_version 指令，因此也将其删除。总而言之，您应删除 app.yaml 的所有部分，以便仅保留 runtime 指令，并指定新版 Python 3，例如 3.10。以下是 app.yaml 在这些更新前后的外观：

之前：

runtime: python27
threadsafe: yes
api_version: 1

handlers:
- url: /.*
  script: main.app

升级后：

runtime: python310

对于尚未准备好从 Python 3 的 app.yaml 中删除所有内容的用户，我们在模块 19 的 Repo 文件夹中提供了一个 app3.yaml 替代文件。如果您希望改用该文件进行部署，请务必将此文件名附加到命令末尾：gcloud app deploy app3.yaml（否则，系统将默认使用您未更改的 Python 2 app.yaml 文件来部署应用）。

appengine_config.py

如果您要升级到 Python 3，则无需使用 appengine_config.py，因此请将其删除。之所以不必指定它们，是因为第三方库支持仅需要在 requirements.txt 中指定它们。Python 2 用户，请继续阅读。

模块 18 appengine_config.py 包含支持第三方库的相应代码，例如 Flask 和刚刚添加到 requirements.txt 的 Cloud 客户端库：

之前：

from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)

不过，仅凭此代码不足以支持刚刚添加的内置库（setuptools、grpcio）。还需要添加几行代码，因此请更新 appengine_config.py，使其如下所示：

升级后：

import pkg_resources
from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)
# Add libraries to pkg_resources working set to find the distribution.
pkg_resources.working_set.add_entry(PATH)

如需详细了解支持 Cloud 客户端库所需的更改，请参阅迁移捆绑服务文档。

其他配置更新

如果您有 lib 文件夹，请将其删除。如果您是 Python 2 用户，请通过发出以下命令来补充 lib 文件夹：

pip install -t lib -r requirements.txt  # or pip2

如果您在开发系统上同时安装了 Python 2 和 3，则可能需要使用 pip2 而不是 pip。

6. 修改应用代码

此部分介绍了对主应用文件 main.py 的更新，将 App Engine Task Queue 拉取队列的使用替换为 Cloud Pub/Sub。Web 模板 templates/index.html 没有变化。这两款应用应以相同的方式运行，并显示相同的数据。

更新导入和初始化

导入和初始化方面有以下几项更新：

对于导入，请将 App Engine NDB 和任务队列替换为 Cloud NDB 和 Pub/Sub。
将 pullq 从 QUEUE 名称重命名为 TOPIC 名称。
对于拉取任务，工作器会租用这些任务一个小时，但对于 Pub/Sub，超时是按消息来衡量的，因此请删除 HOUR 常量。
Cloud API 需要使用 API 客户端，因此请为 Cloud NDB 和 Cloud Pub/Sub 启动这些客户端，其中后者可为主题和订阅提供客户端。
Pub/Sub 需要云项目 ID，因此请从 google.auth.default() 中导入并获取该 ID。
Pub/Sub 要求主题和订阅使用“完全限定的路径名”，因此请使用 *_path() 便利函数创建这些名称。

以下是模块 18 中的导入和初始化，接下来展示了在实现上述更改后，各个部分应如何显示，其中大部分新代码都是各种 Pub/Sub 资源：

之前：

from flask import Flask, render_template, request
from google.appengine.api import taskqueue
from google.appengine.ext import ndb

HOUR = 3600
LIMIT = 10
TASKS = 1000
QNAME = 'pullq'
QUEUE = taskqueue.Queue(QNAME)
app = Flask(__name__)

升级后：

from flask import Flask, render_template, request
import google.auth
from google.cloud import ndb, pubsub

LIMIT = 10
TASKS = 1000
TOPIC = 'pullq'
SBSCR = 'worker'

app = Flask(__name__)
ds_client  = ndb.Client()
ppc_client = pubsub.PublisherClient()
psc_client = pubsub.SubscriberClient()
_, PROJECT_ID = google.auth.default()
TOP_PATH = ppc_client.topic_path(PROJECT_ID, TOPIC)
SUB_PATH = psc_client.subscription_path(PROJECT_ID, SBSCR)

访问数据模型更新

Visit 数据模型不会发生变化。Datastore 访问需要明确使用 Cloud NDB API 客户端上下文管理器 ds_client.context()。在代码中，这意味着您需要在 Python with 块内将 Datastore 调用封装在 store_visit() 和 fetch_visits() 中。此更新与模块 2 中介绍的更新完全相同。

对于 Pub/Sub，最相关的更改是将 Task Queue 拉取任务的入队替换为将 Pub/Sub 消息发布到 pullq 主题。以下是进行这些更新前后的代码：

之前：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)

def store_visit(remote_addr, user_agent):
    'create new Visit in Datastore and queue request to bump visitor count'
    Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()
    QUEUE.add(taskqueue.Task(payload=remote_addr, method='PULL'))

def fetch_visits(limit):
    'get most recent visits'
    return Visit.query().order(-Visit.timestamp).fetch(limit)

升级后：

class Visit(ndb.Model):
    'Visit entity registers visitor IP address & timestamp'
    visitor   = ndb.StringProperty()
    timestamp = ndb.DateTimeProperty(auto_now_add=True)

def store_visit(remote_addr, user_agent):
    'create new Visit in Datastore and queue request to bump visitor count'
    with ds_client.context():
        Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()
    ppc_client.publish(TOP_PATH, remote_addr.encode('utf-8'))

def fetch_visits(limit):
    'get most recent visits'
    with ds_client.context():
        return Visit.query().order(-Visit.timestamp).fetch(limit)

VisitorCount 数据模型更新

VisitorCount 数据模型不会发生变化，fetch_counts() 也是如此，只是将其 Datastore 查询封装在 with 代码块中，如下所示：

之前：

class VisitorCount(ndb.Model):
    visitor = ndb.StringProperty(repeated=False, required=True)
    counter = ndb.IntegerProperty()

def fetch_counts(limit):
    'get top visitors'
    return VisitorCount.query().order(-VisitorCount.counter).fetch(limit)

升级后：

class VisitorCount(ndb.Model):
    visitor = ndb.StringProperty(repeated=False, required=True)
    counter = ndb.IntegerProperty()

def fetch_counts(limit):
    'get top visitors'
    with ds_client.context():
        return VisitorCount.query().order(-VisitorCount.counter).fetch(limit)

更新 worker 代码

工作器代码的更新仅限于将 NDB 替换为 Cloud NDB，并将任务队列替换为 Pub/Sub，但其工作流保持不变。

将 Datastore 调用封装在 Cloud NDB 上下文管理器 with 块中。
任务队列清理涉及从拉取队列中删除所有任务。使用 Pub/Sub 时，“确认 ID”会收集在 acks 中，然后在最后删除/确认。
任务队列拉取任务的租用方式与 Pub/Sub 消息的拉取方式类似。虽然拉取任务的删除是通过任务对象本身完成的，但 Pub/Sub 消息的删除是通过其确认 ID 完成的。
Pub/Sub 消息载荷需要字节（而非 Python 字符串），因此在发布到主题和从主题拉取消息时，分别需要进行一些 UTF-8 编码和解码。

将 log_visitors() 替换为以下更新后的代码，以实现刚刚描述的更改：

之前：

@app.route('/log')
def log_visitors():
    'worker processes recent visitor counts and updates them in Datastore'
    # tally recent visitor counts from queue then delete those tasks
    tallies = {}
    tasks = QUEUE.lease_tasks(HOUR, TASKS)
    for task in tasks:
        visitor = task.payload
        tallies[visitor] = tallies.get(visitor, 0) + 1
    if tasks:
        QUEUE.delete_tasks(tasks)

    # increment those counts in Datastore and return
    for visitor in tallies:
        counter = VisitorCount.query(VisitorCount.visitor == visitor).get()
        if not counter:
            counter = VisitorCount(visitor=visitor, counter=0)
            counter.put()
        counter.counter += tallies[visitor]
        counter.put()
    return 'DONE (with %d task[s] logging %d visitor[s])\r\n' % (
            len(tasks), len(tallies))

升级后：

@app.route('/log')
def log_visitors():
    'worker processes recent visitor counts and updates them in Datastore'
    # tally recent visitor counts from queue then delete those tasks
    tallies = {}
    acks = set()
    rsp = psc_client.pull(subscription=SUB_PATH, max_messages=TASKS)
    msgs = rsp.received_messages
    for rcvd_msg in msgs:
        acks.add(rcvd_msg.ack_id)
        visitor = rcvd_msg.message.data.decode('utf-8')
        tallies[visitor] = tallies.get(visitor, 0) + 1
    if acks:
        psc_client.acknowledge(subscription=SUB_PATH, ack_ids=acks)
    try:
        psc_client.close()
    except AttributeError:  # special handler for grpcio<1.12.0
        pass

    # increment those counts in Datastore and return
    if tallies:
        with ds_client.context():
            for visitor in tallies:
                counter = VisitorCount.query(VisitorCount.visitor == visitor).get()
                if not counter:
                    counter = VisitorCount(visitor=visitor, counter=0)
                    counter.put()
                counter.counter += tallies[visitor]
                counter.put()
    return 'DONE (with %d task[s] logging %d visitor[s])\r\n' % (
            len(msgs), len(tallies))

主应用处理程序 root() 没有变化。HTML 模板文件 templates/index.html 中也不需要进行任何更改，因此这涵盖了所有必要的更新。恭喜您使用 Cloud Pub/Sub 完成了新的第 19 模块应用！

7. 总结/清理

部署应用，以验证其是否按预期运行，以及是否在任何反映的输出中正常运行。同时运行工作器以处理访问者数量。应用验证完成后，执行任何清理步骤，并考虑后续步骤。

部署并验证应用

确保您已创建 pullq 主题和 worker 订阅。如果已完成上述操作，并且您的示例应用已准备就绪，请使用 gcloud app deploy 部署应用。输出应与模块 18 应用的输出相同，只不过您已成功替换整个底层排队机制：

应用的 Web 前端现在会验证应用是否正常运行。虽然应用的这一部分成功查询并显示了热门访问者和最近访问，但请注意，应用会注册此访问，同时创建一个拉取任务，将此访问者添加到总数中。相应任务目前已加入队列，等待处理。

您可以通过 App Engine 后端服务、cron 作业、浏览 /log 或发出命令行 HTTP 请求来执行此操作。以下是使用 curl 调用工作线程代码的执行和输出示例（请替换为您的 PROJECT_ID）：

$ curl https://PROJECT_ID.appspot.com/log
DONE (with 1 task[s] logging 1 visitor[s])

然后，更新后的数量会在您下次访问网站时显示。大功告成！

清理

常规

如果您暂时不想继续操作，建议您停用 App Engine 应用，以免产生结算费用。不过，如果您想进一步测试或实验，App Engine 平台有免费配额，因此只要您不超过该使用层级，就不会产生费用。这是计算费用，但相关 App Engine 服务也可能会产生费用，因此请查看其价格页面了解详情。如果此迁移涉及其他 Cloud 服务，则这些服务会单独计费。在任何一种情况下，如果适用，请参阅下文中的“本 Codelab 特有的问题”部分。

为了完全公开透明，我们在此说明，部署到 Google Cloud 无服务器计算平台（例如 App Engine）会产生少量 build 和存储费用。Cloud Build 和 Cloud Storage 都有各自的免费配额。存储该图片会占用部分配额。不过，您可能居住在没有此类免费层的地区，因此请注意存储空间用量，以尽可能减少潜在费用。您应查看的特定 Cloud Storage“文件夹”包括：

console.cloud.google.com/storage/browser/LOC.artifacts.PROJECT_ID.appspot.com/containers/images
console.cloud.google.com/storage/browser/staging.PROJECT_ID.appspot.com
上述存储链接取决于您的 PROJECT_ID 和 *LOC*ation，例如，如果您的应用托管在美国，则为“us”。

另一方面，如果您不打算继续学习此应用或其他相关迁移 Codelab，并且想要彻底删除所有内容，请关闭您的项目。

此 Codelab 特有的

以下列出的服务是此 Codelab 特有的。如需了解详情，请参阅各个产品的文档：

Cloud Pub/Sub 的不同组件都有免费层级；请确定您的总体使用量，以便更好地了解费用影响，并参阅其价格页面了解更多详情。
App Engine Datastore 服务由 Cloud Datastore（Datastore 模式的 Cloud Firestore）提供，后者也提供免费层级；如需了解详情，请参阅其价格页面。

后续步骤

除了本教程之外，还有其他迁移模块重点介绍如何从旧版捆绑服务迁出，您可以考虑使用这些模块，包括：

模块 2：从 App Engine ndb 迁移到 Cloud NDB
模块 7-9：从 App Engine 任务队列（推送任务）迁移到 Cloud Tasks
模块 12-13：从 App Engine Memcache 迁移到 Cloud Memorystore
模块 15-16：从 App Engine Blobstore 迁移到 Cloud Storage

App Engine 不再是 Google Cloud 中唯一的无服务器平台。如果您有一个小型 App Engine 应用或功能有限的应用，并希望将其转换为独立的微服务，或者您希望将单体式应用拆分为多个可重用的组件，那么这些都是考虑迁移到 Cloud Functions 的充分理由。如果容器化已成为应用开发工作流的一部分，尤其是当它包含 CI/CD（持续集成/持续交付或部署）流水线时，请考虑迁移到 Cloud Run。以下模块涵盖了这些使用场景：

从 App Engine 迁移到 Cloud Functions：请参阅模块 11
从 App Engine 迁移到 Cloud Run：请参阅模块 4，了解如何使用 Docker 将应用容器化；或参阅模块 5，了解如何在不使用容器、Docker 知识或 Dockerfile 的情况下完成迁移

您可以选择改用其他无服务器平台，但我们建议您先考虑最适合您的应用和使用情形的选项，然后再进行任何更改。

无论您接下来考虑哪个迁移模块，都可以在开源代码库中访问所有无服务器迁移站内容（Codelab、视频、源代码 [如有]）。该代码库的 README 还提供了有关应考虑哪些迁移以及任何相关的迁移模块“顺序”的指南。

8. 其他资源

下面列出了其他资源，可供开发者进一步探索本迁移模块或相关迁移模块以及相关产品。这包括提供有关此内容的反馈的途径、指向代码的链接，以及您可能会觉得有用的各种文档。

Codelab 问题/反馈

如果您发现本 Codelab 存在任何问题，请先搜索您的问题，然后再提交。用于搜索和创建新问题的链接：

迁移时可参考的资源

下表中提供了指向模块 18（开始）和模块 19（完成）的 Repo 文件夹的链接。

Codelab	Python 2	Python 3
模块 18	代码	（不适用）
模块 19（本 Codelab）	代码	（与 Python 2 相同，除非您已按上述说明更新 app.yaml，否则请使用 app3.yaml）

从 App Engine 任务队列拉取任务迁移到 Cloud Pub/Sub（模块 19）

1. 概览

在接下来的实验中

所需条件

调查问卷

您打算如何使用本教程？

您如何评价使用 Python 的体验？

您如何评价自己在使用 Google Cloud 服务方面的经验水平？

2. 背景

3. 设置/准备工作

1. 设置项目

2. 获取基准示例应用

3. （重新）部署并验证基准应用

4. 启用新的 Google Cloud 服务/API

通过 Cloud 控制台

从命令行

4. 创建 Pub/Sub 资源

App Engine 任务队列（拉取）与 Cloud Pub/Sub 术语对比

通过 Cloud 控制台

从命令行

通过代码（简短的 Python 脚本）

5. 更新配置

删除 queue.yaml

requirements.txt

app.yaml

Python 2

Python 3

appengine_config.py

其他配置更新

6. 修改应用代码

更新导入和初始化

访问数据模型更新

VisitorCount 数据模型更新

更新 worker 代码

7. 总结/清理

部署并验证应用

清理

常规

此 Codelab 特有的

后续步骤

8. 其他资源

Codelab 问题/反馈

迁移时可参考的资源

在线参考

App Engine Task Queue

Cloud Pub/Sub

App Engine NDB 和 Cloud NDB (Datastore)

App Engine 平台

其他云信息

视频

许可