จัดคิวฟังก์ชันด้วย Cloud Tasks


ฟังก์ชันคิวงานใช้ประโยชน์จาก Google Cloud Tasks เพื่อช่วยให้คุณแอปทำงานที่ใช้เวลานาน ใช้ทรัพยากรมาก หรือมีแบนด์วิดท์จำกัดแบบไม่พร้อมกันนอกลำดับการทำงานของแอปพลิเคชันหลัก

เช่น สมมติว่าคุณต้องการสร้างข้อมูลสำรองของไฟล์ภาพชุดใหญ่ที่โฮสต์อยู่ใน API ที่มีขีดจำกัดอัตราคำขอในปัจจุบัน หากต้องการเป็นผู้บริโภคที่มีความรับผิดชอบของ API ดังกล่าว คุณต้องเคารพขีดจำกัดอัตราคำขอ นอกจากนี้ งานประเภทที่ทำงานต่อเนื่องนานๆ นี้ยังอาจเสี่ยงที่จะดำเนินการไม่สำเร็จเนื่องจากหมดเวลาและขีดจํากัดของหน่วยความจํา

หากต้องการลดความซับซ้อนนี้ คุณสามารถเขียนฟังก์ชันคิวงานที่จะตั้งค่าตัวเลือกงานพื้นฐาน เช่น scheduleTime และ dispatchDeadline จากนั้นส่งฟังก์ชันไปยังคิวใน Cloud Tasks สภาพแวดล้อม Cloud Tasks ได้รับการออกแบบมาโดยเฉพาะเพื่อให้การควบคุมความแออัดและนโยบายการลองอีกครั้งมีประสิทธิภาพสําหรับการดำเนินการประเภทเหล่านี้

Firebase SDK สำหรับ Cloud Functions for Firebase v3.20.1 ขึ้นไปทำงานร่วมกับ Firebase Admin SDK v10.2.0 ขึ้นไปเพื่อรองรับฟังก์ชันของคิวงาน

การใช้ฟังก์ชันคิวงานกับ Firebase อาจส่งผลให้มีการเรียกเก็บเงินสำหรับการประมวลผลCloud Tasks ดูข้อมูลเพิ่มเติมเกี่ยวกับราคา Cloud Tasks

สร้างฟังก์ชันคิวงาน

หากต้องการใช้ฟังก์ชันคิวงาน ให้ทําตามเวิร์กโฟลว์นี้

  1. เขียนฟังก์ชันคิวงานโดยใช้ Firebase SDK สําหรับ Cloud Functions
  2. ทดสอบฟังก์ชันโดยทริกเกอร์ด้วยคำขอ HTTP
  3. ติดตั้งใช้งานฟังก์ชันด้วย Firebase CLI เมื่อทำให้ฟังก์ชันคิวงานใช้งานได้เป็นครั้งแรก CLI จะสร้างคิวงานใน Cloud Tasks พร้อมตัวเลือก (การจำกัดอัตราและการลองใหม่) ที่ระบุไว้ในซอร์สโค้ด
  4. เพิ่มงานลงในคิวงานที่สร้างขึ้นใหม่ โดยส่งพารามิเตอร์เพื่อตั้งค่ากำหนดการเรียกใช้ หากจำเป็น ซึ่งทำได้โดยเขียนโค้ดโดยใช้ Admin SDK และนำไปใช้งานใน Cloud Functions for Firebase

เขียนฟังก์ชันคิวงาน

ตัวอย่างโค้ดในส่วนนี้อิงจากแอปที่ตั้งค่าบริการที่สำรองข้อมูลทั้งหมดจากภาพดาราศาสตร์ประจำวันของนาซา ในการเริ่มต้น ให้นำเข้าโมดูลที่จำเป็น:

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");
index.js

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
main.py

ใช้ onTaskDispatched หรือ on_task_dispatched สำหรับฟังก์ชันคิวงาน เมื่อเขียนฟังก์ชันคิวงาน คุณสามารถตั้งค่าการลองอีกครั้งต่อคิวและการกำหนดค่าการจำกัดอัตรา

กำหนดค่าฟังก์ชันคิวงาน

ฟังก์ชันคิวงานมาพร้อมกับการตั้งค่าการกําหนดค่าที่มีประสิทธิภาพชุดหนึ่งเพื่อควบคุมขีดจํากัดอัตราและลักษณะการลองอีกครั้งของคิวงานอย่างแม่นยํา

Node.js

exports.backupapod = onTaskDispatched(
    {
      retryConfig: {
        maxAttempts: 5,
        minBackoffSeconds: 60,
      },
      rateLimits: {
        maxConcurrentDispatches: 6,
      },
    }, async (req) => {
index.js

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."""
main.py

  • retryConfig.maxAttempts=5: ระบบจะลองทำแต่ละงานในคิวงานซ้ำโดยอัตโนมัติสูงสุด 5 ครั้ง วิธีนี้ช่วยบรรเทาข้อผิดพลาดชั่วคราว เช่น ข้อผิดพลาดของเครือข่ายหรือการหยุดชะงักชั่วคราวของบริการภายนอกที่ต้องอาศัย

  • retryConfig.minBackoffSeconds=60: ระบบจะลองทำแต่ละงานอีกครั้งโดยเว้นระยะเวลาอย่างน้อย 60 วินาทีจากแต่ละครั้งที่ลอง วิธีนี้จะช่วยให้มีเวลาพักระหว่างแต่ละครั้งที่พยายามมากพอ เพื่อไม่ให้การพยายาม 5 ครั้งหมดไปอย่างรวดเร็ว

  • rateLimits.maxConcurrentDispatch=6: โดยจะส่งงานพร้อมกันมากที่สุด 6 งาน วิธีนี้ช่วยให้มั่นใจว่าจะมีคำขอสตรีมอย่างต่อเนื่องไปยังฟังก์ชันที่สำคัญ และช่วยลดจำนวนอินสแตนซ์ที่ใช้งานอยู่และ Cold Start

ทดสอบฟังก์ชันคิวงาน

ในกรณีส่วนใหญ่ Cloud Functionsโปรแกรมจำลองเป็นวิธีที่ดีที่สุดในการทดสอบฟังก์ชันคิวงาน ดูเอกสารประกอบของ Emulator Suite เพื่อดูวิธีเครื่องมือวัดแอปสำหรับการจําลองฟังก์ชันคิวงาน

นอกจากนี้ ระบบจะแสดง functions_sdk สำหรับคิวงานเป็นฟังก์ชัน HTTP แบบง่ายใน Firebase Local Emulator Suite คุณสามารถทดสอบฟังก์ชันงานจำลองได้โดยส่งคำขอ HTTP POST ที่มีเพย์โหลดข้อมูล JSON ดังนี้

 # 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 ที่เหมาะสมให้กับผู้ใช้ที่เรียกใช้คำสั่งทำให้ใช้งานได้

จัดคิวฟังก์ชันของคิวงาน

คุณสามารถจัดคิวฟังก์ชันคิวงานใน Cloud Tasks จากสภาพแวดล้อมเซิร์ฟเวอร์ที่เชื่อถือได้ เช่น Cloud Functions for Firebase โดยใช้ Firebase Admin SDK สำหรับ Node.js หรือไลบรารี Google Cloud สำหรับ Python หากคุณเพิ่งเริ่มใช้ 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);
    });
index.js

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")
main.py

  • โค้ดตัวอย่างจะพยายามกระจายการดำเนินการของงานโดยเชื่อมโยงการหน่วงเวลานาทีที่ N กับงาน N ซึ่งหมายความว่าจะทริกเกอร์งานประมาณ 1 งาน/นาที โปรดทราบว่าคุณยังใช้ scheduleTime (Node.js) หรือ schedule_time (Python) ได้ด้วย หากต้องการให้ Cloud Tasks ทริกเกอร์งานในเวลาที่เฉพาะเจาะจง

  • โค้ดตัวอย่างจะกำหนดเวลาสูงสุดที่ Cloud Tasks จะรอให้งานเสร็จสมบูรณ์ Cloud Tasks จะลองทำงานอีกครั้งหลังจากกำหนดค่าการลองอีกครั้งของคิว หรือจนกว่าจะถึงกำหนดเวลานี้ ในตัวอย่าง คิวได้รับการกำหนดค่าให้ลองทำงานใหม่ได้สูงสุด 5 ครั้ง แต่งานจะถูกยกเลิกโดยอัตโนมัติหากกระบวนการทั้งหมด (รวมถึงการพยายามลองอีกครั้ง) ใช้เวลานานกว่า 5 นาที

ดึงข้อมูลและใส่ URI เป้าหมาย

เนื่องด้วยวิธีที่ Cloud Tasks สร้างโทเค็นการตรวจสอบสิทธิ์เพื่อตรวจสอบสิทธิ์คำขอไปยังฟังก์ชันคิวงานที่สำคัญ คุณต้องระบุ URL ของ Cloud Run ของฟังก์ชันเมื่อจัดคิวงาน เราขอแนะนำให้คุณเรียกข้อมูล URL ของฟังก์ชันแบบเป็นโปรแกรมตามที่แสดงด้านล่าง

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;
}
index.js

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
main.py

การแก้ปัญหา

Cloud Tasks">เปิดการบันทึก Cloud Tasks

บันทึกจาก Cloud Tasks มีข้อมูลการวินิจฉัยที่เป็นประโยชน์ เช่น สถานะคำขอที่เชื่อมโยงกับงาน โดยค่าเริ่มต้น ระบบจะปิดบันทึกจาก Cloud Tasks เนื่องจากมีบันทึกจำนวนมากที่อาจสร้างขึ้นในโปรเจ็กต์ เราขอแนะนำให้คุณเปิดบันทึกการแก้ไขข้อบกพร่องขณะพัฒนาและแก้ไขข้อบกพร่องฟังก์ชันคิวงาน ดูหัวข้อการเปิดการบันทึก

สิทธิ์ IAM

คุณอาจเห็นข้อผิดพลาด PERMISSION DENIED เมื่อจัดคิวงานหรือเมื่อ Cloud Tasks พยายามเรียกใช้ฟังก์ชันคิวงาน ตรวจสอบว่าโปรเจ็กต์มีการเชื่อมโยง IAM ต่อไปนี้

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

gcloud functions add-iam-policy-binding $FUNCTION_NAME \
  --region=us-central1 \
  --member=serviceAccount:${PROJECT_ID}@appspot.gserviceaccount.com \
  --role=roles/cloudfunctions.invoker