此页面由 Cloud Translation API 翻译。

如何在 Flask 应用中使用 App Engine 任务队列（推送任务）（模块 7）

1. 概览

无服务器迁移站系列 Codelab（自定进度的实操教程）和相关视频旨在指导 Google Cloud 无服务器开发者完成一次或多次迁移（主要是从旧服务迁移），从而实现应用的现代化改造。这样做可以提高应用的可移植性，为您提供更多选择和灵活性，让您能够集成并访问更广泛的 Cloud 产品，并更轻松地升级到较新版本。虽然本系列最初主要面向的是最早接触 Cloud 的用户（主要是 App Engine（标准环境）开发者），但涵盖的范围非常广泛，涵盖了 Cloud Functions 和 Cloud Run 等其他无服务器平台，或其他无服务器平台（如适用）。

此 Codelab 会教您如何在模块 1 Codelab 的示例应用中使用 App Engine 任务队列推送任务。第 7 单元博文和视频是对本教程的补充，简要介绍本教程中的内容。

在本单元中，我们将添加使用 push 任务，然后将该用法迁移到第 8 单元及后续单元中的 Cloud Tasks，并迁移到第 9 单元中的 Python 3 和 Cloud Datastore。使用任务队列处理“拉取”任务的用户将迁移到 Cloud Pub/Sub，并应参阅模块 18-19。

在接下来的实验中

使用 App Engine 任务队列 API/捆绑服务
向基本 Python 2 Flask App Engine NDB 应用添加了推送任务用法

所需条件

具有活跃 GCP 结算账号的 Google Cloud 项目
基本 Python 技能
常用 Linux 命令的实践知识
具备开发和部署 App Engine 应用的基础知识
一个可正常运行的模块 1 App Engine 应用（建议完成相应 Codelab [推荐] 或从代码库中复制应用）

调查问卷

您将如何使用本教程？

仅阅读教程内容

阅读并完成练习

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

新手水平

中等水平

熟练水平

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

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

中级

熟练

2. 背景

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

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

迁移模块 18-19 介绍了拉取任务迁移，而模块 7-9 则侧重于推送任务迁移。如需从 App Engine 任务队列推送任务进行迁移，请将其添加到通过模块 1 Codelab 生成的现有 Flask 和 App Engine NDB 应用中。在该应用中，一次新的网页浏览会记录一次新的访问，并向用户显示最近一次的访问。由于旧的访问不会再显示并占用 Datastore 中的空间，因此我们将创建一个推送任务来自动删除最早的访问。在单元 8 中，我们会先将该应用从任务队列迁移到 Cloud Tasks。

本教程包含以下步骤：

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

3. 设置/准备工作

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

设置 Cloud 项目
获取基准示例应用
（重新）部署并验证基准应用

这些步骤可确保您从有效的代码开始。

1. 设置项目

如果您已完成第 1 单元 Codelab，我们建议您重复使用同一项目（和代码）。或者，您也可以创建一个全新的项目或重复使用其他现有项目。确保项目具有有效的结算账号并且已启用 App Engine。

2. 获取基准示例应用

完成此 Codelab 的前提条件之一是具有有效的模块 1 App Engine 应用：完成模块 1 Codelab（推荐），或从代码库中复制模块 1 应用。无论您使用我们的代码还是我们的代码，我们都将从第 1 单元的代码开始。此 Codelab 将逐步引导您完成每个步骤，最后使用与模块 7 代码库文件夹“FINISH”中的代码类似的代码。

START：Module 1 folder (Python 2)
完成：Module 7 folder (Python 2)
整个代码库（用于克隆或下载 ZIP 文件）

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

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

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

执行以下步骤（重新）部署模块 1 应用：

删除 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 部署示例应用
确认模块 1 应用按预期运行，而不显示最近的访问记录（如下图所示）

4. 更新配置

无需对标准 App Engine 配置文件（app.yaml、requirements.txt、appengine_config.py）进行任何更改。

5. 修改应用文件

主要应用文件是 main.py，本部分中的所有更新都与该文件有关。此外，我们还对 Web 模板 templates/index.html 进行了一项次要更新。以下是要在本部分中实现的变更：

更新导入作业
添加推送任务
添加任务处理程序
更新网站模板

1. 更新导入作业

导入 google.appengine.api.taskqueue 会引入任务队列功能。此外，还需要一些 Python 标准库软件包：

由于我们要添加的任务来删除最早的访问，因此应用需要处理时间戳，这意味着要使用 time 和 datetime。
为了记录有关任务执行的实用信息，我们需要 logging。

添加所有这些导入后，您的代码在这些更改前后如下所示：

之前：

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

之后：

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

2. 添加推送任务（整理任务数据、将新任务排入队列）

推送队列文档指出：“要处理任务，您必须将其添加到推送队列。App Engine 提供了一个名为 default 的默认推送队列，该队列已配置完毕，可以使用默认设置进行使用。如果愿意，您可以将所有任务添加到默认队列，而不必创建和配置其他队列。"为简洁起见，此 Codelab 使用 default 队列。如需详细了解如何定义具有相同或不同特征的推送队列，请参阅创建推送队列文档。

此 Codelab 的主要目标是添加一个任务（向 default 推送队列），该任务的任务是删除 Datastore 中不再显示的旧访问。基准应用通过创建新的 Visit 实体来注册每次访问（对 / 的 GET 请求），然后提取并显示最近的访问。所有最早的访问都不会再显示或使用，因此推送任务会删除时间早于显示的访问时间的所有访问。为此，应用的行为需要略微更改：

查询最近的访问时，请不要立即返回这些访问，而应修改应用以保存上次显示时间最早的 Visit 的时间戳，可以放心删除早于此时间的所有访问。
创建一个使用此时间戳作为载荷的推送任务，并将其定向到可通过 HTTP POST 访问 /trim 的任务处理程序。具体而言，使用标准 Python 实用程序转换数据存储区时间戳并将其（以浮点数形式）发送到任务，但也记录时（以字符串形式）并将该字符串作为标记值返回以显示给用户。

所有这些都在 fetch_visits() 中进行，具体如下：

之前：

def fetch_visits(limit):
    return (v.to_dict() for v in Visit.query().order(
            -Visit.timestamp).fetch(limit))

之后：

def fetch_visits(limit):
    'get most recent visits and add task to delete older visits'
    data = Visit.query().order(-Visit.timestamp).fetch(limit)
    oldest = time.mktime(data[-1].timestamp.timetuple())
    oldest_str = time.ctime(oldest)
    logging.info('Delete entities older than %s' % oldest_str)
    taskqueue.add(url='/trim', params={'oldest': oldest})
    return (v.to_dict() for v in data), oldest_str

3. 添加任务处理程序（任务运行时调用的代码）

虽然在 fetch_visits() 中可以很容易地删除旧的访问记录，但请注意，此功能与最终用户无关。它是一项辅助功能，非常适合在标准应用请求之外进行异步处理。最终用户将受益于更快的查询速度，因为 Datastore 中的信息更少。创建一个新函数 trim()，通过任务队列 POST 请求对 /trim 进行调用，该函数将执行以下操作：

提取“最早的访问”时间戳载荷
发出 Datastore 查询以查找早于该时间戳的所有实体。
选择运行更快的“仅限键”因为不需要实际的用户数据。
记录要删除的实体数（包括零）。
调用 ndb.delete_multi() 以删除任何实体（否则会跳过）。
返回一个空字符串（以及隐式 HTTP 200 返回代码）。

您可以在下面的trim()中查看全部信息。将其添加到 main.py，使其紧跟在 fetch_visits() 之后：

@app.route('/trim', methods=['POST'])
def trim():
    '(push) task queue handler to delete oldest visits'
    oldest = request.form.get('oldest', type=float)
    keys = Visit.query(
            Visit.timestamp < datetime.fromtimestamp(oldest)
    ).fetch(keys_only=True)
    nkeys = len(keys)
    if nkeys:
        logging.info('Deleting %d entities: %s' % (
                nkeys, ', '.join(str(k.id()) for k in keys)))
        ndb.delete_multi(keys)
    else:
        logging.info('No entities older than: %s' % time.ctime(oldest))
    return ''   # need to return SOME string w/200

4. 更新网站模板

使用此 Jinja2 条件更新 Web 模板 templates/index.html，以显示最早的时间戳（如果该变量存在）：

{% if oldest is defined %}
    <b>Deleting visits older than:</b> {{ oldest }}</p>
{% endif %}

将以下代码段添加到显示的访问列表之后、结束正文之前，使您的模板如下所示：

<!doctype html>
<html>
<head>
<title>VisitMe Example</title>
<body>

<h1>VisitMe example</h1>
<h3>Last 10 visits</h3>
<ul>
{% for visit in visits %}
    <li>{{ visit.timestamp.ctime() }} from {{ visit.visitor }}</li>
{% endfor %}
</ul>

{% if oldest is defined %}
    <b>Deleting visits older than:</b> {{ oldest }}</p>
{% endif %}
</body>
</html>

6. 摘要/清理

在此 Codelab 的最后，本部分将部署应用，验证应用是否按预期以及任何反映的输出中正常运行。应用验证完成后，请执行所有清理操作，并考虑后续步骤。

部署并验证应用

使用 gcloud app deploy 部署应用。输出内容应与模块 1 中的应用相同，只不过底部有一行新代码，显示了将删除的访问：

4aa8a2cb5f527079

恭喜您完成此 Codelab。您的代码现在应与 Module 7 repo folder 中的内容一致。现在，您可以在单元 8 中迁移到 Cloud Tasks。

清理

常规

如果您目前已完成，我们建议您停用 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 独有的服务。有关详情，请参阅各个产品的文档：

根据旧版捆绑服务（例如任务队列）的价格页面，App Engine 任务队列服务不会产生任何额外费用。
App Engine Datastore 服务由 Cloud Datastore（Datastore 模式的 Cloud Firestore）提供，该服务也有一个免费层级；如需了解详情，请参阅其价格页面。

后续步骤

在此“迁移”中您向模块 1 示例应用添加了任务队列推送队列使用情况，添加了对访问者跟踪的支持，从而产生了模块 7 示例应用。下一次迁移会教您如何从 App Engine 推送任务升级到 Cloud Tasks（如果您愿意的话）。从 2021 年秋季开始，用户在升级到 Python 3 时不再需要迁移到 Cloud Tasks。如需了解详情，请参阅下一部分。

如果您确实想改用 Cloud Tasks，请参阅第 8 单元 Codelab。除此之外，还需要考虑其他迁移，例如 Cloud Datastore、Cloud Memorystore、Cloud Storage 或 Cloud Pub/Sub（拉取队列）。还可以跨产品迁移到 Cloud Run 和 Cloud Functions。所有 Serverless Migration Station 内容（Codelab、视频和源代码 [若有]）均可通过其开源代码库访问。

7. 迁移到 Python 3

2021 年秋季，App Engine 团队将对许多捆绑服务的支持扩展到了第 2 代运行时（最初仅在第 1 代运行时中提供），这意味着您在将应用移植到 Python 3 时，不再需要从 App Engine 任务队列等捆绑服务迁移到独立的 Cloud 或第三方等效服务（例如 Cloud Tasks）。换句话说，只要您修改代码以从下一代运行时访问捆绑服务，就可以继续在 Python 3 App Engine 应用中使用任务队列。

如需详细了解如何将捆绑式服务迁移至 Python 3，请观看第 17 单元 Codelab 及其相应视频。虽然该主题不在模块 7 的讨论范围内，但以下链接提供了已移植到 Python 3 并且仍在使用 App Engine NDB 和任务队列的模块 1 和 7 应用的 Python 3 版本。

8. 其他资源

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

Codelab 问题/反馈

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

迁移时可参考的资源

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

Codelab	Python 2	Python 3
模块 1	代码	code（本教程中未提供）
第 7 单元（此 Codelab）	代码	code（本教程中未提供）