此页面由 Cloud Translation API 翻译。

从 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。此外，还提供了从 App Engine NDB 到 Cloud NDB 以访问 Datastore 的隐式迁移（主要在模块 2 中介绍），并升级到 Python 3。

在第 18 单元中，您将学习如何在应用中使用拉取任务。在本单元中，您将学习完成第 18 单元的应用，并将该应用迁移到 Cloud Pub/Sub。使用任务队列执行推送任务的用户将改用 Cloud Tasks，并应参阅模块 7-9。

在接下来的实验中

将 App Engine 任务队列（拉取任务）替换为 Cloud Pub/Sub
将 App Engine NDB 替换为 Cloud NDB（另请参阅第 2 单元）
将应用移植到 Python 3

所需条件

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

调查问卷

您打算如何使用本教程？

<ph type="x-smartling-placeholder"></ph> 仅仔细阅读

阅读并完成练习

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

新手水平

中等水平

熟练水平

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

<ph type="x-smartling-placeholder"></ph> 新手

中级

熟练

2. 背景

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

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

迁移模块 7-9 介绍推送任务迁移，而模块 18-19 重点介绍拉取任务迁移。虽然 Cloud Tasks 与任务队列推送任务匹配得更密切，但 Pub/Sub 与任务队列拉取任务并不类似。

与任务队列提供的拉取功能相比，Pub/Sub 具有更多功能。例如，Pub/Sub 还具有推送功能，但 Cloud Tasks 更像是任务队列推送任务，因此任何迁移模块都不涵盖 Pub/Sub 推送。第 19 单元 Codelab 演示了如何将队列机制从任务队列拉取队列切换到 Pub/Sub，以及如何从 App Engine NDB 迁移到 Cloud NDB 以使用 Datastore，同时重复执行模块 2 迁移。

虽然第 18 单元的代码作为 Python 2 示例应用，该源代码本身与 Python 2 和 3 兼容，即使在模块 19 中迁移到 Cloud Pub/Sub（和 Cloud NDB）后，它也依然如此。

本教程包含以下步骤：

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

3. 设置/准备工作

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

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

以下步骤可确保您从有效的代码入手，并且已准备好迁移到 Cloud 服务。

1. 设置项目

如果您已完成第 18 单元 Codelab，请重复使用同一项目（和代码）。或者，创建一个全新的项目或重复使用其他现有项目。确保项目具有有效的结算账号和已启用的 App Engine 应用。在本 Codelab 中，您需要找到自己的项目 ID，以便在遇到 PROJECT_ID 变量时使用该 ID。

2. 获取基准示例应用

前提条件之一是一个有效的模块 18 App Engine 应用，因此请完成其 Codelab（推荐；链接见上文），或从代码库中复制模块 18 代码。无论您使用的是我们自己的解决方案还是我们的解决方案，我们都将从这里开始（“START”）。此 Codelab 将逐步引导您完成迁移，最后使用与模块 19 代码库文件夹（“FINISH”）中的代码类似的代码。

START：Module 18 folder (Python 2)
完成：Module 19 folder（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 和 Python 3，则可能需要改用 pip2。
确保您已安装并初始化 gcloud 命令行工具，并查看了其使用情况。
（可选）如果您不想在发出的每个 gcloud 命令中输入 PROJECT_ID，请使用 gcloud config set project PROJECT_ID 设置您的 Cloud 项目。
使用 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 都具有“始终免费”的特性Tier 配额，只要您不超出这些限制，完成本教程就不会产生费用。Cloud API 可通过 Cloud 控制台或命令行启用，具体取决于您的偏好。

在 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 的库页面底部找到它们。例如，观察 pubsub.googleapis.com 为“Service name”。

完成这些步骤后，您的项目将能够访问这些 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 时，数据会进入“主题”。
排队的数据单元 ：使用任务队列的拉取任务在 Pub/Sub 中称为消息。
数据处理方：借助任务队列，工作器可以访问拉取任务；使用 Pub/Sub，您需要订阅/订阅者才能接收消息
数据提取 ：租用拉取任务等同于（通过订阅）从主题中拉取消息。
清理/完成 ：在拉取队列中删除任务队列任务类似于确认 Pub/Sub 消息

虽然产品队列发生了变化，但工作流程仍然相对相似：

该应用使用名为 pullq 的主题，而不是拉取队列。
应用不会向拉取队列添加任务，而是向主题 (pullq) 发送消息。
名为 worker 的订阅者会从 pullq 主题中拉取消息，而不是工作器从拉取队列中租用任务。
应用会处理消息载荷，从而增加 Datastore 中的访问者计数。
应用会确认已处理的消息，而不是从拉取队列中删除任务。

对于任务队列，设置涉及创建拉取队列。使用 Pub/Sub 时，需要创建主题和订阅才能进行设置。在第 18 单元中，我们在应用执行之外处理了 queue.yaml；同样，对于 Pub/Sub，同样需要执行相同的操作

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

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

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

在 Cloud 控制台中

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

前往 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 创建主题和订阅。使用 pullq 的 TOPIC_ID 和 worker 的 SUBSCRIPTION_ID 执行这些操作会为项目 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 部分以及库版本号（也称为“最新”）了解 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 和 Python 3，则可能需要使用 pip2 而不是 pip。

6. 修改应用代码

本部分介绍对主应用文件 main.py 的更新，以 Cloud Pub/Sub 取代使用 App Engine 任务队列拉取队列。网站模板“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 需要 Cloud 项目 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()。在代码中，这意味着您将 Datastore 调用同时封装在 Python with 块内的 store_visit() 和 fetch_visits() 中。此更新与第 2 单元中介绍的更新相同。

对于 Pub/Sub，最相关的更改是用向 pullq 主题发布 Pub/Sub 消息来取代任务队列拉取任务的排队。下面是进行这些更新之前和之后的代码：

之前：

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 数据模型更新

除了将其 Datastore 查询封装在 with 块内外，VisitorCount 数据模型不会更改，并且会执行 fetch_counts()，如下图所示：

之前：

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)

更新工作器代码

工作器代码更新至将 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. 摘要/清理

部署您的应用，以验证它是否按预期运行并在任何反射输出中正常运行。同时运行 worker 以处理访问者计数。应用验证后，执行所有清理步骤并考虑后续步骤。

部署并验证应用

确保您已创建 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 的具体说明”部分。

为了全面披露，部署到像 App Engine 这样的 Google Cloud 无服务器计算平台会产生少量构建和存储费用。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*格式，例如“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 Blob 存储区迁移到 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 的情况下实现容器化

您可以自行选择是否切换到其他无服务器平台，我们建议您先考虑最适合您的应用和用例的方案，然后再做任何更改。

无论您接下来考虑使用哪种迁移模块，都可以通过其开源代码库访问所有 Serverless Migration Station 内容（Codelab、视频、源代码 [如果有]）。代码库的 README 还提供了有关应考虑哪些迁移以及任何相关“顺序”的指南。迁移模块

8. 其他资源

下面列出的其他资源可帮助开发者进一步探索此迁移模块或相关迁移模块及相关产品。这包括提供有关此内容的反馈、代码链接以及各种可能对您有用的文档的地方。

Codelab 问题/反馈

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

迁移时可参考的资源

下表列出了模块 18（START）和模块 19 (FINISH) 对应的代码库文件夹的链接。

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 数据模型更新

更新工作器代码

7. 摘要/清理

部署并验证应用

清理

常规

此 Codelab 的具体内容

后续步骤

8. 其他资源

Codelab 问题/反馈

迁移时可参考的资源

在线参考

App Engine 任务队列

Cloud Pub/Sub

App Engine NDB 和 Cloud NDB (Datastore)

App Engine 平台

其他 Cloud 信息

视频

许可