Lease delle attività di pull

Una volta che le attività si trovano in una coda in modalità pull, un worker può eseguirne la lease. Dopo l'elaborazione, il worker deve eliminarle.

Prima di iniziare

Contesto importante

  • Questo metodo è applicabile solo ai worker in esecuzione all'interno di un servizio nell'ambiente standard.
  • Quando utilizzi le code pull, sei responsabile della scalabilità dei worker in base al volume di elaborazione.

Lease delle attività

Una volta che le attività si trovano nella coda, un worker può eseguirne la lease di una o più utilizzando il lease_tasks() metodo. Potrebbe esserci un breve ritardo prima che le attività aggiunte di recente utilizzando add() diventino disponibili tramite lease_tasks() .

Quando richiedi una lease, devi specificare il numero di attività di cui eseguire la lease (fino a un massimo di 1000) e la durata della lease in secondi (fino a un massimo di una settimana). La durata della lease deve essere sufficientemente lunga da garantire che l'attività più lenta abbia il tempo di terminare prima della scadenza del periodo di lease. Puoi modificare una lease di attività utilizzando modify_task_lease() .

La lease di un'attività la rende non disponibile per l'elaborazione da parte di un altro worker e rimane non disponibile fino alla scadenza della lease.

Il metodo lease_tasks() restituisce un oggetto Task contenente un elenco di attività di cui è stata eseguita la lease dalla coda.

Il seguente esempio di codice esegue la lease di 100 attività dalla coda pull-queue per un'ora:

from google.appengine.api import taskqueue

q = taskqueue.Queue('pull-queue')
q.lease_tasks(3600, 100)

Elaborazione in batch con i tag delle attività

Non tutte le attività sono uguali. Il codice può "taggare" le attività e poi scegliere le attività di cui eseguire la lease in base al tag. Il tag funge da filtro.

from google.appengine.api import taskqueue

q = taskqueue.Queue('pull-queue')
q.add(taskqueue.Task(payload='parse1', method='PULL', tag='parse'))
q.add(taskqueue.Task(payload='parse2', method='PULL', tag='parse'))
q.add(taskqueue.Task(payload='render1', method='PULL', tag='render'))
q.add(taskqueue.Task(payload='render2', method='PULL', tag='render'))

q.lease_tasks_by_tag(3600, 100, 'render') # leases render tasks, but not parse

q.lease_tasks_by_tag(3600, 100) # Leases up to 100 tasks that have same tag.

Regolare le frequenze di polling

I worker che eseguono il polling della coda per le attività di cui eseguire la lease devono rilevare se stanno tentando di eseguire la lease delle attività più velocemente di quanto la coda possa fornirle. Se si verifica questo errore, è possibile generare le seguenti eccezioni da lease_tasks() :

  • `google.appengine.api.taskqueue.TransientError`
  • `google.appengine.runtime.apiproxy_errors.DeadlineExceededError`


Il codice deve rilevare queste eccezioni, eseguire il backoff dalla chiamata a lease_tasks() , e riprovare in un secondo momento. Per evitare questo problema, valuta la possibilità di impostare una scadenza RPC più elevata quando chiami lease_tasks() . Dovresti anche eseguire il backoff quando una richiesta di lease restituisce un elenco vuoto di attività.

Se generi più di 10 richieste LeaseTasks al secondo per coda, solo le prime 10 richieste restituiranno risultati. Se le richieste superano questo limite, viene restituito OK senza risultati.

Monitorare le attività nella Google Cloud consolle

Per visualizzare informazioni su tutte le attività e le code nella tua applicazione:

  1. Apri la pagina Cloud Tasks nella Google Cloud console e cerca il valore Pull nella colonna Type.

    Vai a Cloud Tasks

  2. Fai clic sul nome della coda che ti interessa per aprire la pagina dei dettagli della coda. Vengono visualizzate tutte le attività nella coda selezionata.

Eliminare le attività

Una volta completata un'attività, un worker deve eliminarla dalla coda. Se vedi attività rimanenti in una coda dopo che un worker ha terminato di elaborarle, è probabile che il worker non sia riuscito. In questo caso, le attività verranno elaborate da un altro worker.

Puoi eliminare un elenco di attività, ad esempio quello restituito da lease_tasks() , semplicemente passandolo a delete_tasks():

from google.appengine.api import taskqueue

q = taskqueue.Queue('pull-queue')
tasks = q.lease_tasks(3600, 100)
# Perform some work with the tasks here
q.delete_tasks(tasks)

Un esempio end-to-end di code pull

Per un esempio end-to-end semplice ma completo dell'utilizzo delle code pull in Python, consulta appengine-pullqueue-counter.