工作佇列函式可利用 Google Cloud Tasks 的優勢，在主要應用程式流程之外，以非同步方式執行耗時、耗用大量資源或頻寬受限的工作。
舉例來說，假設您想為大量圖片檔案建立備份，而這些檔案目前託管於設有速率限制的 API 上。為了成為該 API 的負責任使用者，您必須遵守其頻率限制。此外，這類長時間執行的工作可能會因逾時和記憶體限制而發生失敗。
為降低這種複雜性，您可以編寫任務佇列函式，設定 scheduleTime 和 dispatchDeadline 等基本任務選項，然後將函式交給 Cloud Tasks 中的佇列。Cloud Tasks 環境是專門為這類作業設計，可確保有效的壅塞控制和重試政策。
Cloud Functions for Firebase 專用的 Firebase SDK 3.20.1 以上版本與 Firebase Admin SDK 10.2.0 以上版本互通，可支援工作佇列功能。
使用 Firebase 的 Taskqueue 函式可能會產生 Cloud Tasks 處理費用。詳情請參閱 Cloud Tasks 定價。
建立工作佇列函式
如要使用工作佇列函式，請按照下列工作流程操作：
- 使用 Cloud Functions 的 Firebase SDK 編寫工作佇列函式。
- 使用 HTTP 要求觸發函式來測試函式。
- 使用 Firebase CLI 部署函式。首次部署工作佇列函式時，CLI 會在 Cloud Tasks 中建立工作佇列，並使用原始碼中指定的選項 (速率限制和重試)。
- 將工作新增至新建的工作佇列，並傳遞參數，以便視需要設定執行時程。您可以使用 Admin SDK 編寫程式碼，然後部署至 Cloud Functions for Firebase。
編寫工作佇列函式
本節的程式碼範例是以應用程式為基礎，該應用程式會設定服務，以便備份 NASA Astronomy Picture of the Day 的所有圖片。首先，請匯入必要模組：
Node.js
// Dependencies for task queue functions.
const {onTaskDispatched} = require("firebase-functions/v2/tasks");
const {onRequest, HttpsError} = require("firebase-functions/v2/https");
const {getFunctions} = require("firebase-admin/functions");
const {logger} = require("firebase-functions/v2");
// Dependencies for image backup.
const path = require("path");
const fetch = require("node-fetch");
const {initializeApp} = require("firebase-admin/app");
const {getStorage} = require("firebase-admin/storage");
const {GoogleAuth} = require("google-auth-library");
Python
# Dependencies for task queue functions.
from google.cloud import tasks_v2
import requests
from firebase_functions.options import RetryConfig, RateLimits, SupportedRegion
# Dependencies for image backup.
from datetime import datetime, timedelta
import json
import pathlib
from urllib.parse import urlparse
from firebase_admin import initialize_app, storage, functions
from firebase_functions import https_fn, tasks_fn, params
import google.auth
from google.auth.transport.requests import AuthorizedSession
請使用 onTaskDispatched 或 on_task_dispatched 執行工作佇列函式。編寫工作佇列函式時，您可以設定個別佇列的重試和速率限制設定。
設定工作佇列函式
工作佇列功能提供一組強大的設定，可精確控制工作佇列的速率限制和重試行為：
Node.js
exports.backupapod = onTaskDispatched(
{
retryConfig: {
maxAttempts: 5,
minBackoffSeconds: 60,
},
rateLimits: {
maxConcurrentDispatches: 6,
},
}, async (req) => {
Python
@tasks_fn.on_task_dispatched(retry_config=RetryConfig(max_attempts=5, min_backoff_seconds=60),
rate_limits=RateLimits(max_concurrent_dispatches=10))
def backupapod(req: tasks_fn.CallableRequest) -> str:
"""Grabs Astronomy Photo of the Day (APOD) using NASA's API."""
retryConfig.maxAttempts=5：工作佇列中的每項工作最多會自動重試 5 次。這有助於減少暫時性錯誤，例如網路錯誤或依附外部服務的暫時性服務中斷。retryConfig.minBackoffSeconds=60：每項工作會在每項嘗試之間重試，且重試間隔至少為 60 秒。這樣一來，每次嘗試之間就會有較大的緩衝區，我們就不會太快就耗盡 5 次重試嘗試次數。rateLimits.maxConcurrentDispatch=6：在特定時間點最多會調度 6 項工作。這有助於確保對基礎函式的要求穩定流動，並有助於減少執行中的例項和冷啟動次數。
測試工作佇列函式
在大多數情況下，Cloud Functions 模擬器是測試工作佇列函式的最佳方式。請參閱 Emulator Suite 說明文件，瞭解如何為應用程式進行工作佇列函式模擬。
此外，工作佇列 functions_sdk 會在 Firebase Local Emulator Suite 中公開為簡單的 HTTP 函式。您可以傳送含有 JSON 資料酬載的 HTTP POST 要求，藉此測試模擬工作函式：
# start the Local Emulator Suite
firebase emulators:start
# trigger the emulated task queue function
curl \
-X POST # An HTTP POST request...
-H "content-type: application/json" \ # ... with a JSON body
http://localhost:$PORT/$PROJECT_ID/$REGION/$NAME \ # ... to function url
-d '{"data": { ... some data .... }}' # ... with JSON encoded data
部署工作佇列函式
使用 Firebase CLI 部署工作佇列函式：
$ firebase deploy --only functions:backupapod
第一次部署工作佇列函式時，CLI 會在 Cloud Tasks 中建立工作佇列，並使用您在原始碼中指定的選項 (速率限制和重試)。
如果在部署函式時遇到權限錯誤，請務必將適當的 IAM 角色指派給執行部署指令的使用者。
將工作佇列函式排入佇列
您可以使用 Node.js 的 Firebase Admin SDK 或 Python 的 Google Cloud 程式庫，從信任的伺服器環境 (例如 Cloud Functions for Firebase) 將工作佇列函式排入 Cloud Tasks。如果您是 Admin SDK 新手，請參閱「將 Firebase 新增至伺服器」一文，瞭解如何開始使用。
一般流程會建立新工作、在 Cloud Tasks 中排入工作佇列，並設定工作的設定：
Node.js
exports.enqueuebackuptasks = onRequest(
async (_request, response) => {
const queue = getFunctions().taskQueue("backupapod");
const targetUri = await getFunctionUrl("backupapod");
const enqueues = [];
for (let i = 0; i <= BACKUP_COUNT; i += 1) {
const iteration = Math.floor(i / HOURLY_BATCH_SIZE);
// Delay each batch by N * hour
const scheduleDelaySeconds = iteration * (60 * 60);
const backupDate = new Date(BACKUP_START_DATE);
backupDate.setDate(BACKUP_START_DATE.getDate() + i);
// Extract just the date portion (YYYY-MM-DD) as string.
const date = backupDate.toISOString().substring(0, 10);
enqueues.push(
queue.enqueue({date}, {
scheduleDelaySeconds,
dispatchDeadlineSeconds: 60 * 5, // 5 minutes
uri: targetUri,
}),
);
}
await Promise.all(enqueues);
response.sendStatus(200);
});
Python
@https_fn.on_request()
def enqueuebackuptasks(_: https_fn.Request) -> https_fn.Response:
"""Adds backup tasks to a Cloud Tasks queue."""
task_queue = functions.task_queue("backupapod")
target_uri = get_function_url("backupapod")
for i in range(BACKUP_COUNT):
batch = i // HOURLY_BATCH_SIZE
# Delay each batch by N hours
schedule_delay = timedelta(hours=batch)
schedule_time = datetime.now() + schedule_delay
dispatch_deadline_seconds = 60 * 5 # 5 minutes
backup_date = BACKUP_START_DATE + timedelta(days=i)
body = {"data": {"date": backup_date.isoformat()[:10]}}
task_options = functions.TaskOptions(schedule_time=schedule_time,
dispatch_deadline_seconds=dispatch_deadline_seconds,
uri=target_uri)
task_queue.enqueue(body, task_options)
return https_fn.Response(status=200, response=f"Enqueued {BACKUP_COUNT} tasks")
程式碼範例會為第 N 項工作建立 N 分鐘的延遲時間，藉此分散工作執行作業。這表示每分鐘觸發約 1 項工作。請注意，如果您希望 Cloud Tasks 在特定時間觸發工作，也可以使用
scheduleTime(Node.js) 或schedule_time(Python)。範例程式碼會設定 Cloud Tasks 等待工作完成的時間上限。Cloud Tasks 會根據佇列的重試設定重試工作，或重試至期限為止。在範例中，佇列會設定重試工作最多 5 次，但如果整個程序 (包括重試嘗試) 超過 5 分鐘，系統會自動取消工作。
擷取並納入目標 URI
由於 Cloud Tasks 會建立驗證權杖，以便驗證對基礎工作佇列函式的要求，因此您必須在排入工作時指定函式的 Cloud Run 網址。建議您以程式輔助方式擷取函式的網址，如以下所示：
Node.js
/**
* Get the URL of a given v2 cloud function.
*
* @param {string} name the function's name
* @param {string} location the function's location
* @return {Promise<string>} The URL of the function
*/
async function getFunctionUrl(name, location="us-central1") {
if (!auth) {
auth = new GoogleAuth({
scopes: "https://www.googleapis.com/auth/cloud-platform",
});
}
const projectId = await auth.getProjectId();
const url = "https://cloudfunctions.googleapis.com/v2beta/" +
`projects/${projectId}/locations/${location}/functions/${name}`;
const client = await auth.getClient();
const res = await client.request({url});
const uri = res.data?.serviceConfig?.uri;
if (!uri) {
throw new Error(`Unable to retreive uri for function at ${url}`);
}
return uri;
}
Python
def get_function_url(name: str, location: str = SupportedRegion.US_CENTRAL1) -> str:
"""Get the URL of a given v2 cloud function.
Params:
name: the function's name
location: the function's location
Returns: The URL of the function
"""
credentials, project_id = google.auth.default(
scopes=["https://www.googleapis.com/auth/cloud-platform"])
authed_session = AuthorizedSession(credentials)
url = ("https://cloudfunctions.googleapis.com/v2beta/" +
f"projects/{project_id}/locations/{location}/functions/{name}")
response = authed_session.get(url)
data = response.json()
function_url = data["serviceConfig"]["uri"]
return function_url
疑難排解
Cloud Tasks">啟用 Cloud Tasks 記錄
Cloud Tasks 的記錄包含實用的診斷資訊，例如與工作相關聯的請求狀態。由於 Cloud Tasks 可能會在您的專案中產生大量記錄，因此預設情況下會關閉 Cloud Tasks 的記錄。建議您在積極開發及偵錯工作佇列函式時，開啟偵錯記錄。請參閱「開啟記錄功能」。
身分與存取權管理權限
排入工作佇列時，或 Cloud Tasks 嘗試叫用工作佇列函式時，您可能會看到 PERMISSION DENIED 錯誤。請確認專案具有下列身分與存取權管理繫結：
用於將工作排入 Cloud Tasks 佇列的身分需要
cloudtasks.tasks.createIAM 權限。在本範例中，這是 App Engine 的預設服務帳戶
gcloud projects add-iam-policy-binding $PROJECT_ID \
--member=serviceAccount:${PROJECT_ID}@appspot.gserviceaccount.com \
--role=roles/cloudtasks.enqueuer
用於將工作排入 Cloud Tasks 的使用者身分，必須具備使用 Cloud Tasks 中與工作相關聯的服務帳戶的權限。
在本範例中，這是 App Engine 預設服務帳戶。
如要瞭解如何將 App Engine 預設服務帳戶新增為 App Engine 預設服務帳戶的使用者，請參閱 Google Cloud IAM 說明文件。
用於觸發工作佇列函式的身分需要具備
cloudfunctions.functions.invoke權限。在本範例中，這是 App Engine 的預設服務帳戶
gcloud functions add-iam-policy-binding $FUNCTION_NAME \
--region=us-central1 \
--member=serviceAccount:${PROJECT_ID}@appspot.gserviceaccount.com \
--role=roles/cloudfunctions.invoker