จัดคิวฟังก์ชันด้วย 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 {requiresRole} = require("firebase-functions");
const {getFunctions} = require("firebase-admin/functions");
const {logger} = require("firebase-functions");

// Dependencies for image backup.
const {URL, URLSearchParams} = require("node:url");
const path = require("path");
const {initializeApp} = require("firebase-admin/app");
const {getStorage} = require("firebase-admin/storage");

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 เป็นวิธีที่ดีที่สุดในการทดสอบฟังก์ชันคิวงาน ดูเอกสารประกอบของชุดโปรแกรมจำลองเพื่อดูวิธีใช้เครื่องมือวัดแอปสำหรับการจำลองฟังก์ชันคิวงาน

นอกจากนี้ ฟังก์ชันคิวงานจะแสดงเป็นฟังก์ชัน 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 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
            }),
        );
      }
      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 พยายามเรียกใช้ฟังก์ชันคิวงาน ตรวจสอบว่าโปรเจ็กต์มีการเชื่อมโยง 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