如何在 Flask 应用中使用 App Engine 任务队列（拉取任务）（模块 18）

1. 概览

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

此 Codelab 会教您如何在模块 1 Codelab 中的示例应用中添加和使用 App Engine 任务队列拉取任务。我们将在第 18 单元的教程中添加拉取任务的用法，然后在第 19 单元中提前将其迁移到 Cloud Pub/Sub。使用任务队列执行推送任务的用户将改用 Cloud Tasks，并应参阅模块 7-9。

未使用任务队列？

如果您的应用不使用 App Engine 任务队列服务，请跳过模块 7-9（推送任务）和 18-19（拉取任务），或完成这些 Codelab 作为练习，以熟悉任务队列迁移。

在接下来的实验中

• 使用 App Engine 任务队列 API/捆绑服务
• 向基本的 Python 2 Flask App Engine NDB 应用添加拉取队列用法

所需条件

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

调查问卷

您将如何使用本教程？

仅阅读教程内容 阅读并完成练习

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

新手水平 中等水平 熟练水平

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

<ph type="x-smartling-placeholder"></ph> 新手 中级 熟练

2. 背景

如需从 App Engine 任务队列拉取任务进行迁移，请将其添加到通过模块 1 Codelab 生成的现有 Flask 和 App Engine NDB 应用中。示例应用会显示最终用户的最近访问记录。没关系，但更有趣的是，对访问者进行跟踪，以了解哪些访问者访问最多。

尽管我们可以使用推送任务统计这些访问者数量，但我们希望将职责划分为：负责记录访问并立即响应用户的示例应用，以及指定的“Worker”其工作就是在正常的请求-响应工作流之外统计访问者计数。

为了实现此设计，我们要在主应用中添加拉取队列功能并支持工作器功能。工作器可以作为单独的进程（如后端实例或在始终启动的虚拟机上运行的代码）、Cron 作业运行，也可作为使用 curl 或 wget 的基本命令行 HTTP 请求运行。完成此集成后，您可以在下一个（模块 19）Codelab 中将应用迁移到 Cloud Pub/Sub。

本教程包含以下步骤：

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

3. 设置/准备工作

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

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

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

1. 设置项目

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

2. 获取基准示例应用

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

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

无论您使用哪种模块 1 应用，该文件夹都应如以下输出结果所示，可能还包含 lib 文件夹：

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

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

请执行以下步骤，部署模块 1 应用：

1. 删除 lib 文件夹（如有），并运行 pip install -t lib -r requirements.txt 以重新填充 lib。如果您同时安装了 Python 2 和 Python 3，则可能需要改用 pip2 命令。
2. 确保您已安装并初始化 gcloud 命令行工具并查看其使用情况。
3. 如果您不希望在发出每个 gcloud 命令时都输入 PROJECT_ID，请使用 gcloud config set project PROJECT_ID 设置您的 Cloud 项目。
4. 使用 gcloud app deploy 部署示例应用
5. 确认模块 1 应用按预期运行，并显示最近的访问记录（如下图所示）

4. 更新配置

无需对标准 App Engine 配置文件（app.yaml、requirements.txt、appengine_config.py）进行任何更改。而应添加一个新的配置文件 queue.yaml，其中包含以下内容，并将其放入同一顶级目录：

queue:
- name: pullq
  mode: pull

queue.yaml 文件指定应用存在的所有任务队列（App Engine 自动创建的 default [推送] 队列除外）。在本例中，只有一个拉取队列，即名为 pullq 的拉取队列。App Engine 要求将 mode 指令指定为 pull，否则它会默认创建推送队列。如需详细了解如何创建拉取队列，请参阅文档。如需了解其他选项，另请参阅 queue.yaml 参考页面。

请将此文件与您的应用分开部署。您仍将使用 gcloud app deploy，但要在命令行中提供 queue.yaml：

$ gcloud app deploy queue.yaml
Configurations to update:

descriptor:      [/tmp/mod18-gaepull/queue.yaml]
type:            [task queues]
target project:  [my-project]

WARNING: Caution: You are updating queue configuration. This will override any changes performed using 'gcloud tasks'. More details at
https://cloud.google.com/tasks/docs/queue-yaml

Do you want to continue (Y/n)?

Updating config [queue]...⠹WARNING: We are using the App Engine app location (us-central1) as the default location. Please use the "--location" flag if you want to use a different location.
Updating config [queue]...done.

Task queues have been updated.

Visit the Cloud Platform Console Task Queues page to view your queues and cron jobs.
$

5. 修改应用代码

本部分对以下文件进行了更新：

• main.py - 向主应用添加拉取队列
• templates/index.html - 更新 Web 模板以显示新数据

导入和常量

第一步是添加一项新的 import 语句和若干常量来支持拉取队列：

• 添加任务队列库的导入操作 google.appengine.api.taskqueue。
• 添加三个常量，以支持从拉取队列 (QUEUE) 租用一个小时 (HOUR) 的最大拉取任务 (TASKS)。
• 添加一个常量，用于显示最近的访问和最多访问者 (LIMIT)。

以下是原始代码以及进行这些更新后的内容：

之前：

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

app = Flask(__name__)

之后：

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__)

添加拉取任务（收集任务数据并在拉取队列中创建任务）

数据模型 Visit 与查询要在 fetch_visits() 中显示的访问次数相同。这部分代码所需的唯一更改是在 store_visit() 中。除了注册访问之外，还要使用访问者的 IP 地址向拉取队列添加一个任务，以便工作器可以递增访问者计数器。

之前：

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 entity in Datastore'
    Visit(visitor='{}: {}'.format(remote_addr, user_agent)).put()

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'
    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)

创建用于跟踪访问者的数据模型和查询函数

添加数据模型 VisitorCount 来跟踪访问者；它应包含 visitor 本身的字段以及用于跟踪访问次数的整数 counter。然后，添加一个名为 fetch_counts() 的新函数（或者，它可以是 Python classmethod），以查询并返回排名靠前的访问者（按从大到小的顺序）。在 fetch_visits() 正文的正下方添加类和函数：

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

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

添加工作器代码

添加新函数 log_visitors()，以通过 GET 请求将访问者记录到 /log。它使用字典/哈希来跟踪最近的访问者数量，尽可能多地租用一个小时的任务。对于每项任务，它都会统计同一访问者的所有访问。有了统计数据，应用就会更新 Datastore 中已经存在的所有对应的 VisitorCount 实体，或者根据需要创建新的实体。最后一步会返回一条纯文本消息，说明有多少访问者通过已处理的任务数量注册。将此函数添加到 fetch_counts() 正下方的 main.py 中：

@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))

使用新的显示数据更新主处理程序

如需显示名列前茅的访问者，请更新主处理程序 root() 以调用 fetch_counts()。此外，该模板也会进行更新，显示最常访问者的数量和最近的访问次数。将访问者计数以及来自 fetch_visits() 调用的最近访问打包到一个 context 中，以传递给网站模板。以下是进行此项更改前后的代码：

之前：

@app.route('/')
def root():
    'main application (GET) handler'
    store_visit(request.remote_addr, request.user_agent)
    visits = fetch_visits(10)
    return render_template('index.html', visits=visits)

之后：

@app.route('/')
def root():
    'main application (GET) handler'
    store_visit(request.remote_addr, request.user_agent)
    context = {
        'limit':  LIMIT,
        'visits': fetch_visits(LIMIT),
        'counts': fetch_counts(LIMIT),
    }
    return render_template('index.html', **context)

这些是 main.py 所需的全部更改。下面以图示的形式展示了一些更新，仅作说明之用，方便您大致了解对 main.py 做出的更改：

使用新的展示数据更新网站模板

网页模板“templates/index.html”需要进行更新，以便显示最常访问者，而不仅仅是最近访问者的正常载荷。将排名靠前的访问者及其数量放入页面顶部的表格中，并继续像以前一样显示最近的访问。唯一的其他更改是通过 limit 变量指定显示的数字，而不是硬编码数字。以下是您应该对网页模板进行的更新：

之前：

<!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>

之后：

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

<h1>VisitMe example</h1>

<h3>Top {{ limit }} visitors</h3>
<table border=1 cellspacing=0 cellpadding=2>
    <tr><th>Visitor</th><th>Visits</th></tr>
{% for count in counts %}
    <tr><td>{{ count.visitor|e }}</td><td align="center">{{ count.counter }}</td></tr>
{% endfor %}
</table>

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

以上是必要的更改，将 App Engine 任务队列拉入任务添加到模块 1 示例应用中。现在，您的目录代表模块 18 示例应用，并且应包含以下文件：

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

6. 摘要/清理

在此 Codelab 的最后，本部分将部署应用，验证应用是否按预期以及任何反映的输出中正常运行。单独运行工作器以处理访问者计数。应用验证后，执行所有清理步骤并考虑后续步骤。

部署并验证应用

确保您已按照我们在此 Codelab 顶部附近使用 gcloud app deploy queue.yaml 的操作设置了拉取队列。如果该操作已完成，并且您的示例应用已准备就绪，请使用 gcloud app deploy 部署您的应用。输出应该与模块 1 应用相同，只不过现在该应用包含“热门访问者”表格：

虽然更新后的网络前端会显示热门访问者和最近的访问，但请注意，访问者计数不包括此次访问。应用程序显示之前的访问者计数，同时丢弃一个新任务，使该访问者在拉入队列中的计数递增，该任务正在等待处理。

您可以通过多种方式调用 /log 来执行任务：

• App Engine 后端服务
• cron 作业
• 网络浏览器
• 命令行 HTTP 请求（curl、wget 等）

例如，如果您使用 curl 向 /log 发送 GET 请求，且您提供了 PROJECT_ID，则输出将如下所示：

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

更新后的数量会反映在下一次网站访问中。大功告成！

恭喜您完成此 Codelab，并成功将 App Engine 任务队列拉取队列服务添加到示例应用。现在，您可以在模块 19 中迁移到 Cloud Pub/Sub、Cloud NDB 和 Python 3。

清理

常规

如果您目前已完成，我们建议您停用 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 的示例应用中，从而实现了模块 18 的示例应用。在下一次迁移中，您会将 App Engine 拉取任务升级到 Cloud Pub/Sub。从 2021 年底开始，用户在升级到 Python 3 时不再需要迁移到 Cloud Pub/Sub。如需了解详情，请参阅下一部分。

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

7. 迁移到 Python 3

2021 年秋季，App Engine 团队将对许多捆绑服务的支持扩展到了第 2 代运行时（具有第 1 代运行时）。因此，您在将应用移植到 Python 3 时不再需要从捆绑服务（例如 App Engine 任务队列）迁移到独立的云端或第三方服务（例如 Cloud Pub/Sub）。换句话说，只要您修改代码以从下一代运行时访问捆绑服务，就可以继续在 Python 3 App Engine 应用中使用任务队列。

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

8. 其他资源

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

Codelab 问题/反馈

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

• 搜索问题
• 提交新问题/反馈

迁移时可参考的资源

下表列出了模块 1（START）和模块 18 (FINISH) 对应的代码库文件夹的链接。您也可以从所有 App Engine Codelab 迁移的代码库中访问这些库；克隆该代码库或下载一个 ZIP 文件。

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

在线参考

以下是与本教程相关的资源：

App Engine 任务队列

• App Engine 任务队列概览
• App Engine 任务队列拉取队列概览
• App Engine 任务队列拉取队列完整示例应用
• 创建任务队列拉取队列
• 2011 年 Google I/O 大会拉取队列启动视频（votelator 示例应用）
• queue.yaml 参考文档
• queue.yaml 与 Cloud Tasks
• 将拉取队列拉取到 Pub/Sub 迁移指南
• 将 App Engine 任务队列拉取队列到 Cloud Pub/Sub 的文档示例

App Engine 平台

App Engine 文档

Python 2 App Engine（标准环境）运行时

• 在 Python 2 App Engine 上使用 App Engine 内置库

Python 3 App Engine（标准环境）运行时

Python 2 与3 个 App Engine（标准环境）运行时

Python 2 到 3 App Engine（标准环境）迁移指南

App Engine 价格和配额信息

第二代 App Engine 平台发布（2018 年）

对旧版运行时的长期支持

文档迁移示例

• 社区提供的迁移示例

其他 Cloud 信息

• 在 Google Cloud Platform 上使用 Python 应用
• Google Cloud Python 客户端库
• Google Cloud“始终免费”计划层级
• Google Cloud SDK（gcloud 命令行工具）
• 所有 Google Cloud 文档

视频

• 无服务器迁移站
• 无服务器实境教学
• 订阅 Google Cloud Tech
• 订阅 Google Developers

许可

此作品已获得 Creative Commons Attribution 2.0 通用许可授权。