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

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

เช่น สมมติว่าคุณต้องการสร้างข้อมูลสำรองของชุดไฟล์รูปภาพจำนวนมากที่โฮสต์อยู่ใน API ที่มีขีดจำกัดอัตรา คุณต้องเคารพขีดจำกัดอัตราการใช้งานของ 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. เขียนฟังก์ชันคิวงานโดยใช้ SDK ของ Firebase สำหรับ Cloud Functions
  2. ทดสอบฟังก์ชันโดยทริกเกอร์ด้วยคำขอ HTTP
  3. ติดตั้งใช้งานฟังก์ชันด้วย Firebase CLI เมื่อทําการติดตั้งใช้งานฟังก์ชันคิวงานเป็นครั้งแรก CLI จะสร้างคิวงานใน Cloud Tasks พร้อมตัวเลือก (การจำกัดอัตราคำขอและการลองใหม่) ที่ระบุไว้ในซอร์สโค้ด
  4. เพิ่มงานลงในคิวงานที่สร้างขึ้นใหม่ โดยส่งพารามิเตอร์เพื่อตั้งค่า กำหนดเวลาการดำเนินการหากจำเป็น คุณสามารถทำได้โดยเขียนโค้ด โดยใช้ Admin SDK และนำไปใช้งานใน Cloud Functions for Firebase

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

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

Node.js

// Dependencies for task queue functions.
const {onTaskDispatched} = require("firebase-functions/tasks");
const {onRequest, HttpsError} = require("firebase-functions/https");
const {getFunctions} = require("firebase-admin/functions");
const {logger} = require("firebase-functions");

// Dependencies for image backup.
const path = require("path");
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 งานในเวลาที่กำหนด ซึ่งจะช่วยให้มั่นใจได้ว่าจะมีคำขอไปยังฟังก์ชันพื้นฐานอย่างต่อเนื่อง และช่วยลดจำนวนอินสแตนซ์ที่ใช้งานอยู่และ Cold Start

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

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

นอกจากนี้ ฟังก์ชันคิวงานยังแสดงเป็นฟังก์ชัน 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);
    });

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 งาน/นาที โปรดทราบว่าคุณยังใช้ scheduleTime (Node.js) หรือ schedule_time (Python) ได้ด้วยหากต้องการ Cloud Tasksเรียกใช้ทาสก์ในเวลาที่เฉพาะเจาะจง

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

การแก้ปัญหา

ดูวิธีวินิจฉัยปัญหาเกี่ยวกับคิวงานโดยใช้การบันทึก และกำหนดค่า สิทธิ์ IAM ที่จำเป็น

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

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

สิทธิ์ IAM

คุณอาจเห็นข้อผิดพลาด PERMISSION DENIED เมื่อจัดคิวงานหรือเมื่อ Cloud Tasks พยายามเรียกใช้ฟังก์ชันคิวงาน ตรวจสอบว่าโปรเจ็กต์มี Binding ของ 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บัญชีบริการเริ่มต้น

ดูเอกสารประกอบเกี่ยวกับ Google Cloud IAM เพื่อดูวิธีการเพิ่มApp Engineบัญชีบริการเริ่มต้น เป็นผู้ใช้App Engineบัญชีบริการเริ่มต้น

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