העברת תורי הודעות מסוג Push ל-Cloud Tasks ‏ (Python)

בדף הזה מוסבר איך אפשר להעביר קוד של תור דחיפה (push queue) מ-Task Queues ל-Cloud Tasks. ‫Cloud Tasks היא הדרך המומלצת לעבודה עם תורים של הודעות Push ב-App Engine.

באמצעות Cloud Tasks, אתם ניגשים לאותו שירות שאליו אתם ניגשים באמצעות Task Queues RPC API. המשמעות היא שלא צריך ליצור מחדש את תורי ההודעות הקיימים ואת משימות ה-Push. עם זאת, כדי להשתמש ב-Cloud Tasks API, צריך להעביר קוד שיוצר תורים של הודעות בדחיפה או משימות בדחיפה, או קוד שמבצע אינטראקציה עם התורים או המשימות האלה.

אפשר ליצור תורים של משימות מסוג push ומשימות מסוג push ולבצע איתם אינטראקציה באמצעות ממשקי ה-API ל-REST ול-RPC של Cloud Tasks, ספריית הלקוח של Cloud Tasks, Google Cloud CLI ומסוףGoogle Cloud . בדף הזה מובאות דוגמאות לשימוש ב-gcloud CLI ובספריית הלקוח של Cloud Tasks.

ב-Cloud Tasks, כל התורים פועלים כתורי push. בהמשך המדריך הזה ובמסמכי Cloud Tasks, המונח תור זהה למונח תור דחיפה. באופן דומה, המונח משימה שווה ערך למונח משימת push.

תכונות שלא זמינות ב-Cloud Tasks

התכונות הבאות לא זמינות ב-Cloud Tasks:

  • הוספת משימות לתור בעסקאות ב-Datastore
  • שימוש בספריית המשימות שמושהות במקום בשירות worker
  • עבודה עם משימות באפליקציות לפי עקרון ה-Multi-tenancy
  • סימולציה באמצעות שרת הפיתוח המקומי
  • הוספת משימות באופן אסינכרוני

תמחור ומכסות

העברה של תורי הודעות בדחיפה ל-Cloud Tasks עשויה להשפיע על התמחור והמכסות של האפליקציה.

תמחור

ל-Cloud Tasks יש תמחור משלו. בדומה לתורי משימות, שליחת בקשות לאפליקציית App Engine עם משימה עלולה לגרום לאפליקציה לצבור עלויות.

מכסות

המכסות של Cloud Tasks שונות מהמכסות של תורי משימות. בדומה ל-Task Queues, שליחת בקשות לאפליקציית App Engine מ-Cloud Tasks עשויה להשפיע על מכסות הבקשות של App Engine.

לפני ההעברה

אם עדיין לא עשיתם זאת, צריך להגדיר את סביבת הפיתוח של Python כך שתשתמש בגרסת Python שתואמת ל- Google Cloud, ולהתקין כלי בדיקה ליצירת סביבות Python מבודדות.

בקטעים הבאים מפורטים שלבי ההגדרה לפני העברת תורי ההודעות מסוג Push ל-Cloud Tasks.

העברת תורים להוצאת הודעות

לפני שמתחילים, צריך להעביר תורים של שליפת הודעות לפני שמבצעים את ההוראות במדריך הזה להעברת תורים של דחיפת הודעות. לא מומלץ להעביר תורי משיכה אחרי העברת תורי דחיפה, כי סביר להניח שהשימוש הנדרש בקובץ queue.yaml יגרום להתנהגות לא צפויה ב-Cloud Tasks.

הגנה על הגדרות התור

אחרי שמתחילים את תהליך ההעברה ל-Cloud Tasks, שינוי הקובץ queue.yaml עלול לגרום להתנהגות לא צפויה ולא מומלץ. כדי להגן על הגדרות התור מפני שינויים בקובץ queue.yaml, מבצעים את השלבים הבאים.

  1. מגדירים את ה-CLI של gcloud כך שקובץ queue.yaml לא ייכלל בהמשך הפריסות.

    מוסיפים את קובץ queue.yaml לקובץ .gcloudignore. כדי לבדוק אם כבר יש לכם קובץ .gcloudignore, אתם יכולים להריץ את הפקודה הבאה בטרמינל מהספרייה ברמה העליונה של האפליקציה. הפקודה הזו תציג את שם הקובץ אם הוא קיים.

    ls -a | grep .gcloudignore

    מידע נוסף על קובצי .gcloudignore זמין במאמר בנושא קובצי עזר של .gcloudignore.

  2. מגבילים את ההרשאות בקובץ queue.yaml.

    מומלץ לפעול לפי השיטות המומלצות שמתוארות במדריך שלנו בנושא הגדרת אבטחה לתור.

  3. מידע על Cloud Tasks ועל קובץ queue.yaml (אופציונלי).

    כשמשתמשים ב-Cloud Tasks API כדי לנהל את הגדרות התור, פריסה של קובץ queue.yaml מבטלת את ההגדרות שנקבעו על ידי Cloud Tasks, וזה עלול לגרום להתנהגות בלתי צפויה. מידע נוסף זמין במאמר שימוש בניהול תורים לעומת queue.yaml.

הפעלת Cloud Tasks API

כדי להפעיל את Cloud Tasks API, לוחצים על Enable ב-Cloud Tasks API ב-API Library. אם מופיע לחצן Manage (ניהול) במקום לחצן Enable (הפעלה), סימן שהפעלתם בעבר את Cloud Tasks API בפרויקט, ואין צורך להפעיל אותו שוב.

אימות האפליקציה ב-Cloud Tasks API

צריך לאמת את האפליקציה ב-Cloud Tasks API. בקטע הזה נדון באימות בשני תרחישים שונים לדוגמה.

כדי לפתח או לבדוק את האפליקציה באופן מקומי, מומלץ להשתמש בחשבון שירות. הוראות להגדרת חשבון שירות ולחיבור שלו לאפליקציה מופיעות במאמר קבלת פרטי כניסה לחשבון שירות וצירופם באופן ידני.

כדי לפרוס את האפליקציה ב-App Engine, לא צריך לספק אימות חדש. פרטי האימות של אפליקציות App Engine מוסקים מ-Application Default Credentials‏ (ADC).

הורדת ה-CLI של gcloud

הורידו והתקינו את ה-CLI של gcloud כדי להשתמש בו עם Cloud Tasks API אם לא התקנתם אותו בעבר. אם ה-CLI של gcloud כבר מותקן, מריצים את הפקודה הבאה מהטרמינל.

gcloud components update

ייבוא ספריות לקוח של Cloud

כדי להשתמש בספריית הלקוח של Cloud Tasks באפליקציית App Engine:

  1. מעדכנים את הקובץ app.yaml. פועלים לפי ההוראות שמתאימות לגרסת Python:

    ‫Python 2

    באפליקציות Python 2, מוסיפים את הגרסאות האחרונות של הספריות grpcio ו-setuptools.

    קובץ app.yaml לדוגמה:

    runtime: python27
threadsafe: yes
api_version: 1

libraries:
- name: grpcio
  version: latest
- name: setuptools
  version: latest

    Python 3

    באפליקציות Python 3, מציינים את הרכיב runtime בקובץ app.yaml עם גרסה נתמכת של Python 3. לדוגמה:

    runtime: python310 # or another support version

    סביבת זמן הריצה של Python 3 מתקינה ספריות באופן אוטומטי, כך שלא צריך לציין ספריות מובנות מסביבת זמן הריצה הקודמת של Python 2. אם האפליקציה שלכם ב-Python 3 משתמשת בשירותים אחרים מדור קודם שכלולים בחבילה במהלך ההעברה, אתם יכולים להמשיך לציין את הספריות המובנות הנדרשות. אחרת, אפשר למחוק את השורות המיותרות בקובץ app.yaml.

  2. מעדכנים את הקובץ requirements.txt. פועלים לפי ההוראות שמתאימות לגרסת Python:

    ‫Python 2

    מוסיפים את ספריות הלקוח של Cloud ל-Cloud Tasks לרשימת יחסי התלות בקובץ requirements.txt.

    google-cloud-tasks

    לאחר מכן מריצים את הפקודה pip install -t lib -r requirements.txt כדי לעדכן את רשימת הספריות שזמינות לאפליקציה.

    Python 3

    מוסיפים את ספריות הלקוח של Cloud ל-Cloud Tasks לרשימת יחסי התלות בקובץ requirements.txt.

    google-cloud-tasks

    ‫App Engine מתקין באופן אוטומטי את התלות האלה במהלך פריסת האפליקציה בסביבת זמן הריצה של Python 3, לכן צריך למחוק את התיקייה lib אם היא קיימת.

  3. באפליקציות Python 2, אם האפליקציה משתמשת בספריות מובנות או בספריות שהועתקו, צריך לציין את הנתיבים האלה בקובץ appengine_config.py שנמצא באותה תיקייה שבה נמצא קובץ app.yaml:

    import pkg_resources
from google.appengine.ext import vendor

# Set PATH to your libraries folder.
PATH = 'lib'
# Add libraries installed in the PATH folder.
vendor.add(PATH)
# Add libraries to pkg_resources working set to find the distribution.
pkg_resources.working_set.add_entry(PATH)

    הקובץ appengine_config.py מניח שספריית העבודה הנוכחית היא המקום שבו נמצאת התיקייה lib. במקרים מסוימים, כמו בדיקות יחידה, ספריית העבודה הנוכחית יכולה להיות שונה. כדי להימנע משגיאות, אפשר להעביר באופן מפורש את הנתיב המלא לתיקייה lib באמצעות:

    import os
path = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'lib')

  4. מייבאים את ספריית הלקוח של Cloud Tasks לכל הקבצים שמשתמשים בתורי Push מ-Task Queues API:

    from google.cloud import tasks

    אחרי שתשלימו את ההעברה של כל הקוד שיוצר תורים של הודעות בדחיפה או מבצע אינטראקציה איתם אל Cloud Tasks, תצטרכו להסיר את ההצהרות שמייבאות את Task Queues API. לדוגמה, from google.appengine.api import taskqueue.

יצירה וניהול של תורים

בקטע הזה מוסבר איך ליצור ולנהל תורים באמצעות Cloud Tasks API.

ב-Cloud Tasks, לא משתמשים בקובץ queue.yaml כדי ליצור תורים או לנהל אותם. במקום זאת, משתמשים ב-Cloud Tasks API. לא מומלץ להשתמש גם בקובץ queue.yaml וגם ב-Cloud Tasks API, אבל יכול להיות שזה יהיה חלק בלתי נמנע מהמעבר מ-Task Queues ל-Cloud Tasks, בהתאם לאפליקציה שלכם. כדאי לקרוא את המאמר שימוש בניהול תורים לעומת queue.yaml כדי ללמוד על שיטות מומלצות.

יצירת תורים

כדאי לקרוא את הקטע הזה אם האפליקציה שלכם יוצרת תורים באופן פרוגרמטי, או אם אתם רוצים ליצור תורים נוספים משורת הפקודה.

ב-Cloud Tasks, שמות התורים הם בפורמט projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID. החלק LOCATION_ID בשם התור תואם ל Google Cloud אזור. החלק QUEUE_ID בשם התור שווה לשדה name של התור ב-Task Queues. שם התור נוצר במהלך יצירת התור על סמך הפרויקט, האזור וQUEUE_ID שאתם מציינים.

באופן כללי, המיקום של התור (כלומר, האזור) צריך להיות זהה לאזור של האפליקציה. יש שני חריגים לכלל הזה: אפליקציות שמשתמשות באזור europe-west ואפליקציות שמשתמשות באזור us-central. ב-Cloud Tasks, האזורים האלה נקראים europe-west1 ו-us-central1 בהתאמה.

אפשר להגדיר את התור במהלך היצירה שלו, אבל אפשר גם לעדכן אותו אחרי שהוא נוצר.

אין צורך ליצור מחדש תורים קיימים. במקום זאת, אפשר להעביר את הקוד שפועל באינטראקציה עם התורים הקיימים על ידי קריאת החלקים הרלוונטיים במדריך הזה.

שימוש חוזר בשמות של תורים

צריך להמתין 7 ימים אחרי מחיקת תור כדי ליצור תור עם אותו מזהה תור באותו פרויקט ובאותו מיקום (כלומר, באותו אזור).

בדוגמה הבאה נוצרות שתי תורים באמצעות Cloud Tasks. לתור הראשון יש מזהה תור queue-blue, והוא מוגדר לשליחת כל המשימות לגרסה v2 של השירות task-module בקצב של 5/s. לתור השני יש מזהה תור queue-red והוא משבץ משימות בקצב של 1/s. שניהם נוצרים בפרויקט עם מזהה הפרויקט my-project-id במיקום us-central1. זהו המקבילה ב-Cloud Tasks ליצירת תורים ב-Task Queues.

gcloud

ה-CLI של gcloud מסיק את הפרויקט והמיקום מההגדרה של ה-CLI של gcloud.

gcloud tasks queues create queue-blue \
--max-dispatches-per-second=5 \
--routing-override=service:task-module,version:v2
 
gcloud tasks queues create queue-red \
--max-dispatches-per-second=1

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue_blue_name = 'queue-blue'
# queue_red_name = 'queue-red'

parent = f"projects/{project}/locations/{location}"

queue_blue = {
    "name": client.queue_path(project, location, queue_blue_name),
    "rate_limits": {"max_dispatches_per_second": 5},
    "app_engine_routing_override": {"version": "v2", "service": "task-module"},
}

queue_red = {
    "name": client.queue_path(project, location, queue_red_name),
    "rate_limits": {"max_dispatches_per_second": 1},
}

queues = [queue_blue, queue_red]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

מידע נוסף זמין במאמר בנושא יצירת תור של Cloud Tasks.

הגדרת קצב העיבוד של התור

בטבלה הבאה מפורטים השדות ששונים בין Task Queues לבין Cloud Tasks.

השדה Task Queues (תורי משימות) שדה Cloud Tasks תיאור
rate max_dispatches_per_second הקצב המקסימלי שבו המערכת משייכת משימות מהתור
max_concurrent_requests max_concurrent_dispatches המספר המקסימלי של משימות בו-זמניות שאפשר לשלוח מהתור
bucket_size max_burst_size

‫Cloud Tasks מחשב מאפיין לקריאה בלבד max_burst_size שמגביל את מהירות העיבוד של משימות בתור על סמך הערך של max_dispatches_per_second. השדה הזה מאפשר לתור להגיע לקצב גבוה כדי שהעיבוד יתחיל זמן קצר אחרי שהמשימה נוספת לתור, אבל עדיין מגביל את השימוש במשאבים כשמתווספות לתור הרבה משימות בפרק זמן קצר.

בתורים של App Engine שנוצרו או עודכנו באמצעות קובץ queue.yaml, הערך של max_burst_size שווה בהתחלה לערך של bucket_size. עם זאת, אם התור יועבר בהמשך לפקודה update באמצעות ממשק Cloud Tasks כלשהו, הערך של max_burst_size יתאפס על סמך הערך של max_dispatches_per_second, בלי קשר לעדכון של max_dispatches_per_second.

total_storage_limit הוצא משימוש ב-Cloud Tasks ב-Cloud Tasks אין תמיכה בהגדרת מגבלת אחסון מותאמת אישית

אפשר להגדיר את קצב העיבוד של התור כשיוצרים את התור או כשמעדכנים אותו בהמשך. בדוגמה הבאה נעשה שימוש ב-Cloud Tasks כדי להגדיר את קצב העיבוד בתור שנקרא queue-blue שכבר נוצר. אם queue-blue נוצר או הוגדר באמצעות קובץ queue.yaml, הדוגמה הבאה מאפסת את max_burst_size על סמך הערך max_dispatches_per_second של 20. זהו הערך המקביל ב-Cloud Tasks להגדרת קצב העיבוד של התור ב-Task Queues.

gcloud

gcloud tasks queues update queue-blue \
--max-dispatches-per-second=20 \
--max-concurrent-dispatches=10

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue-blue'

# Get queue object
queue_path = client.queue_path(project, location, queue)
queue = client.get_queue(name=queue_path)

# Update queue object
queue.rate_limits.max_dispatches_per_second = 20
queue.rate_limits.max_concurrent_dispatches = 10

response = client.update_queue(queue=queue)
print(response)

מידע נוסף זמין במאמר הגדרת מגבלות קצב.

השבתה והפעלה מחדש של תורים

ב-Cloud Tasks משתמשים במונח השהיה באותו אופן שבו משתמשים במונח השבתה ב-Task Queues. השהיה של תור מפסיקה את הביצוע של המשימות בתור עד להפעלה מחדש של התור. אבל אפשר להמשיך להוסיף משימות לתור שהושהה. ב-Cloud Tasks, המונח resume (המשך) משמש באותו אופן כמו ב-Task Queues.

בדוגמה הבאה מושהה תור עם מזהה התור queue1 . הפעולה הזו מקבילה להשבתה של תורים ב-Task Queues.

gcloud

gcloud tasks queues pause 

queue1

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.pause_queue(name=queue_path)

מידע נוסף זמין במאמר בנושא Cloud Tasks בקטע השהיית תורים.

מחיקת תורים

אחרי שמוחקים תור, צריך לחכות 7 ימים לפני שיוצרים תור עם אותו שם. אם אתם לא יכולים לחכות 7 ימים, כדאי למחוק את כל המשימות מתור ההמתנה ולהגדיר מחדש את תור ההמתנה.

בדוגמה הבאה נמחקת התור עם מזהה התור queue1 . זהו המקבילה ב-Cloud Tasks למחיקת תורים ב-Task Queues.

gcloud

gcloud tasks queues delete 

queue1

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.delete_queue(name=queue_path)

מידע נוסף זמין במאמר בנושא Cloud Tasks מחיקת תורים.

יצירה וניהול של משימות

בקטע הזה מוסבר איך ליצור משימות ולנהל אותן באמצעות Cloud Tasks API.

יצירת משימות

בטבלה הבאה מפורטים השדות ששונים בין Task Queues לבין Cloud Tasks.

השדה Task Queues (תורי משימות) שדה Cloud Tasks תיאור
מה חדש ב-Cloud Tasks app_engine_http_request יוצר בקשה שמיועדת לשירות App Engine. המשימות האלה נקראות משימות של App Engine.
method http_method מציינת את שיטת הבקשה, לדוגמה: POST
url relative_uri מציין את המטפל במשימה. שימו לב להבדל באות האחרונה: i עבור מזהה משאבים אחיד במקום l עבור מען משאבים אחיד
target app_engine_routing זה שינוי אופציונלי. מציינים את service, version ו-instance של App Engine למשימת App Engine. אם לא מגדירים את המאפיין, המערכת משתמשת בגרסת השירות, במופע ובשירות שמוגדרים כברירת מחדל.

בדוגמה הבאה נוצרת משימה שמופנית לשירות App Engine בשם worker עם ה-handler‏ /update_counter. זהו המקבילה ב-Cloud Tasks ליצירת משימות בתורי משימות.

gcloud

gcloud tasks create-app-engine-task --queue=default \
--method=POST --relative-uri=/update_counter --routing=service:worker \
--body-content=10

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
amount = 10

parent = client.queue_path(project, location, queue)

task = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "body": str(amount).encode(),
    }
}

response = client.create_task(parent=parent, task=task)
eta = response.schedule_time.strftime("%m/%d/%Y, %H:%M:%S")
print(f"Task {response.name} enqueued, ETA {eta}.")

מידע נוסף זמין במאמר בנושא יצירת משימות App Engine במאמרי העזרה של Cloud Tasks.

ציון שירות היעד והניתוב

הגדרה של שירות היעד, הגרסה והמופע של App Engine למשימות של App Engine היא אופציונלית. כברירת מחדל, משימות ב-App Engine מנותבות לשירות, לגרסה ולמופע שמוגדרים כברירת מחדל בזמן הניסיון לבצע את המשימה.

במהלך יצירת המשימה, מגדירים את המאפיין app_engine_routing של המשימה כדי לציין שירות, גרסה או מופע אחרים של App Engine עבור המשימה.

כדי להפנות את כל המשימות בתור מסוים לאותו שירות, גרסה ומופע של App Engine, אפשר להגדיר את המאפיין app_engine_routing_override בתור.

מידע נוסף זמין במאמר בנושא Cloud Tasks הגדרת ניתוב.

העברת נתונים אל ה-handler

בדומה לתורי משימות, אפשר להעביר נתונים ל-handler בשתי דרכים באמצעות Cloud Tasks. אפשר להעביר נתונים כפרמטרים של שאילתה ב-URI היחסי, או להעביר נתונים בגוף הבקשה באמצעות שיטות HTTP‏ POST או PUT.

ב-Cloud Tasks, המונח body משמש באותו אופן שבו המונח payload משמש ב-Task Queues. ב-Cloud Tasks, סוג התוכן שמוגדר כברירת מחדל לגוף הבקשה הוא octet-stream ולא טקסט פשוט. אפשר להגדיר את סוג התוכן של גוף ההודעה על ידי ציון שלו בכותרת.

בדוגמה הבאה מועבר מפתח ל-handler‏ /update_counter בשתי דרכים שונות. זהו המקבילה ב-Cloud Tasks להעברת נתונים ל-handler ב-Task Queues.

console

gcloud tasks create-app-engine-task --queue=default --method=GET  \
--relative-uri=

/update_counter

?key=blue --routing=service:worker
 
gcloud tasks create-app-engine-task --queue=default --method=POST \
--relative-uri=

/update_counter

 --routing=service:worker \
--body-content="{'key': 'blue'}"

ספריית משתמש

import json

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'

parent = client.queue_path(project, location, queue)

task1 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter?key=blue",
        "app_engine_routing": {"service": "worker"},
    }
}

task2 = {
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.POST,
        "relative_uri": "/update_counter",
        "app_engine_routing": {"service": "worker"},
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"key": "blue"}).encode(),
    }
}

response = client.create_task(parent=parent, task=task1)
print(response)
response = client.create_task(parent=parent, task=task2)
print(response)

מתן שמות למשימות

אפשר לציין את שם המשימה, אבל זה לא חובה. אם לא מציינים את שם המשימה, Cloud Tasks יוצר אותו בשבילכם על ידי יצירת מזהה משימה והסקת הפרויקט והמיקום (כלומר, האזור) על סמך התור שציינתם במהלך יצירת המשימה.

שמות המשימות הם בפורמט projects/PROJECT_ID/locations/LOCATION_ID/queues/QUEUE_ID/tasks/TASK_ID. החלק TASK_ID בשם המשימה שווה לשדה name של המשימה ב-Task Queues.

שימוש חוזר בשמות של משימות

צריך לחכות לפני שמשתמשים שוב בשם של משימה. משך הזמן שצריך להמתין לפני שמבצעים את הפעולה משתנה בהתאם לשאלה אם תור השליחה של המשימה נוצר ב-Cloud Tasks או ב-Task Queues.

לגבי משימות בתורים שנוצרו באמצעות Task Queues (כולל תור ברירת המחדל), צריך להמתין כ-9 ימים אחרי שהמשימה המקורית נמחקה או בוצעה. לגבי משימות בתורים שנוצרו באמצעות Cloud Tasks, צריך להמתין כשעה אחרי שהמשימה המקורית נמחקה או בוצעה.

בדוגמה הבאה נוצרת משימה עם הערך TASK_ID שמוגדר ל-first-try, והיא מתווספת לתור ברירת המחדל. זהו השם המקביל ב-Cloud Tasks לשם המשימה ב-Task Queues.

gcloud

ה-CLI של gcloud יוצר את שם המשימה על סמך הפרויקט והמיקום שמוגדרים בהגדרות.

gcloud tasks create-app-engine-task first-try --queue=default \
--method=GET --relative-uri=

/url/path

ספריית משתמש

כשמשתמשים בספריית הלקוח, צריך לציין את השם המלא של המשימה אם רוצים לציין את TASK_ID. הפרויקט והמיקום צריכים להיות זהים לפרויקט ולמיקום של התור שאליו המשימה מתווספת.

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'default'
# task_name = 'first-try'

parent = client.queue_path(project, location, queue)

task = {
    "name": client.task_path(project, location, queue, task_name),
    "app_engine_http_request": {
        "http_method": tasks.HttpMethod.GET,
        "relative_uri": "/url/path",
    },
}
response = client.create_task(parent=parent, task=task)
print(response)

ניסיון חוזר של משימות שנכשלו

אפשר להגדיר את ההגדרה של ניסיון חוזר לביצוע משימה בתורים במהלך יצירת התור, או על ידי עדכון התור. בטבלה הבאה מפורטים השדות של Task Queues והשדות התואמים ב-Cloud Tasks.

השדה Task Queues (תורי משימות) שדה Cloud Tasks
task_retry_limit max_attempts
task_age_limit max_retry_duration
min_backoff_seconds min_backoff
max_backoff_seconds max_backoff
max_doublings max_doublings

פרמטרים של ניסיון חוזר ספציפי למשימה

פרמטרים ספציפיים של ניסיון חוזר למשימה שהוגדרו ב-Task Queues פועלים ב-Cloud Tasks, אבל אי אפשר לערוך אותם או להגדיר אותם במשימות חדשות. כדי לשנות את פרמטרים הניסיון החוזר של משימה שיש לה פרמטרים ספציפיים לניסיון חוזר, צריך ליצור מחדש את המשימה עם תור של Cloud Tasks שיש לו את פרמטרים הניסיון החוזר הרצויים.

בדוגמה הבאה מוצגים תרחישי ניסיון חוזר שונים:

  • ב-fooqueue, המערכת מנסה לבצע מחדש משימות עד שבע פעמים, למשך עד יומיים מניסיון ההפעלה הראשון. אחרי שמגיעים לשתי המגבלות, הפעולה נכשלת באופן סופי.
  • ב-barqueue,‏ App Engine מנסה לבצע מחדש משימות, ומגדיל את המרווח באופן לינארי בין כל ניסיון חוזר עד שמגיע להשהיה לפני ניסיון חוזר (backoff) המקסימלית, ואז מנסה שוב ללא הגבלה במרווח המקסימלי (כך שהמרווחים בין הבקשות הם 10 שניות, 20 שניות, 30 שניות וכו', עד 190 שניות, 200 שניות, 200 שניות וכו').
  • בדוגמה bazqueue, מרווח הזמן בין הניסיונות החוזרים מתחיל ב-10 שניות, מוכפל שלוש פעמים, ואז גדל באופן לינארי. לבסוף, המערכת מנסה שוב ושוב ללא הגבלה במרווח המקסימלי (כך שמרווחי הזמן בין הבקשות הם 10 שניות, 20 שניות, 40 שניות, 80 שניות, 160 שניות, 240 שניות, 300 שניות, 300 שניות וכן הלאה).

זהו המקבילה ב-Cloud Tasks לניסיון חוזר של משימות בתורי משימות.

gcloud

כשמגדירים אפשרויות שמציינות מספר שניות, צריך לכלול s אחרי המספר השלם (למשל 200s ולא 200).

gcloud tasks queues create fooqueue \
--max-attempts=7 \
--max-retry-duration=172800s  #2*60*60*24 seconds in 2 days
 
gcloud tasks queues create barqueue \
--min-backoff=10s \
--max-backoff=200s \
--max-doublings=0
 
gcloud tasks queues create bazqueue \
--min-backoff=10s \
--max-backoff=300s \
--max-doublings=3

ספריית משתמש

from google.protobuf import duration_pb2

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# fooqueue = 'fooqueue'
# barqueue = 'barqueue'
# bazqueue = 'bazqueue'

parent = f"projects/{project}/locations/{location}"

max_retry = duration_pb2.Duration()
max_retry.seconds = 2 * 60 * 60 * 24

foo = {
    "name": client.queue_path(project, location, fooqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"max_attempts": 7, "max_retry_duration": max_retry},
}

min = duration_pb2.Duration()
min.seconds = 10

max = duration_pb2.Duration()
max.seconds = 200

bar = {
    "name": client.queue_path(project, location, barqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 0},
}

max.seconds = 300
baz = {
    "name": client.queue_path(project, location, bazqueue),
    "rate_limits": {"max_dispatches_per_second": 1},
    "retry_config": {"min_backoff": min, "max_backoff": max, "max_doublings": 3},
}

queues = [foo, bar, baz]
for queue in queues:
    response = client.create_queue(parent=parent, queue=queue)
    print(response)

מידע נוסף זמין במאמר בנושא Cloud Tasks הגדרת פרמטרים של ניסיון חוזר.

מחיקת משימות מתור

כשמוחקים משימה, צריך לחכות 9 ימים לפני שיוצרים משימה עם אותו שם אם המשימה הייתה בתור שנוצר באמצעות קובץ queue.yaml, או שעה אחת אם המשימה הייתה בתור שנוצר באמצעות Cloud Tasks.

בדוגמה הבאה נמחקת המשימה עם מזהה המשימה foo מהתור עם מזהה התור queue1. זהו המקבילה ב-Cloud Tasks למחיקת משימות בתורי משימות.

gcloud

הפרויקט והמיקום של המשימה נגזרים מפרויקט ברירת המחדל של ה-CLI של gcloud.

gcloud tasks delete foo --queue=queue1

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

task_path = client.task_path(project, location, queue, "foo")
response = client.delete_task(name=task_path)

מידע נוסף זמין במאמר בנושא Cloud Tasks מחיקת משימה מתור.

מחיקת משימות

בדוגמה הבאה, כל המשימות בתור עם מזהה התור queue1 נמחקות. זוהי הפעולה המקבילה ב-Cloud Tasks לניקוי משימות בתורי משימות.

gcloud

הפרויקט והמיקום של התור נגזרים מפרויקט ברירת המחדל של ה-CLI של gcloud.

gcloud tasks queues purge 

queue1

ספריית משתמש

client = tasks.CloudTasksClient()

# TODO(developer): Uncomment these lines and replace with your values.
# project = 'my-project-id'
# location = 'us- central1'
# queue = 'queue1'

queue_path = client.queue_path(project, location, queue)
response = client.purge_queue(name=queue_path)

מידע נוסף זמין במאמר בנושא מחיקת כל המשימות מתור בתיעוד של Cloud Tasks.

המאמרים הבאים