Processi di lavoro delle code

I processi di lavoro delle code elaborano elementi dalle code di lavoro usando schede del browser concorrenti. Ogni processo di lavoro riceve la propria scheda, estrae elementi dalla coda, esegue uno script su ciascun elemento e conferma o annulla il risultato.

Concetto

1. Metti in coda elementi di lavoro (URL, record di dati, attività)
2. I processi di lavoro estraggono gli elementi e li elaborano, ciascuno nella propria scheda del browser
3. Lo script per singolo elemento chiama g.worker.getContext() per ottenere l’elemento, svolge il lavoro, poi chiama g.queue.commit() o g.queue.abort()

Il pannello Processi di lavoro (Workers) mostra pool attivi, conteggi elaborati, errori, profondità della coda e controlli per mettere in pausa o fermare il lavoro.

Due modi per definire i processi di lavoro

Dichiarativo: queues.json

Crea un file queues.json nel percorso principale (root) dello spazio di lavoro:

{
  "pools": [
    {
      "queue": "urls",
      "script": "scripts/process-url.js",
      "concurrency": 3,
      "throttle": 30,
      "layout": "horizontal",
      "clearTabs": true
    }
  ]
}

Campo	Obbligatorio	Predefinito	Descrizione
queue	Sì	—	Nome della coda da elaborare
script	Sì	—	Percorso dello script relativo al percorso principale (root) dello spazio di lavoro
concurrency	No	1	Numero di processi di lavoro concorrenti (schede)
throttle	No	60	Numero massimo di estrazioni al minuto per processo di lavoro
layout	No	—	Disposizione delle schede: horizontal, vertical o quad
clearTabs	No	false	Chiude tutte le schede esistenti prima dell’avvio
requiresWorkflowEnvelope	No	false	Richiede che ogni elemento di coda contenga metadati del registro di tracciabilità dei flussi di lavoro prima dell’esecuzione

Avvia un pool dichiarato dalla console delle code, dal pannello Code compatto o dagli strumenti MCP.

Programmatico: g.workers.start()

async function main() {
  // Metti in coda un po' di lavoro
  for (let i = 1; i <= 100; i++) {
    g.queue.enqueue("pages", {
      url: "https://example.com/page/" + i
    });
  }

  // Avvia i processi di lavoro
  g.workers.start({
    clearTabs: true,
    layout: "horizontal",
    workers: [
      {
        queue: "pages",
        script: "scripts/scrape-page.js",
        concurrency: 4,
        throttle: { requestsPerMinute: 20 },
        requiresWorkflowEnvelope: false
      }
    ]
  });
}

g.workers.start() avvia i processi e prosegue: i processi di lavoro vengono eseguiti indipendentemente dallo script che li ha avviati.

Script per singolo elemento

Ogni processo di lavoro esegue il tuo script una volta per ogni elemento di coda. Lo script deve:

1. Chiamare g.worker.getContext() per ottenere l’elemento e l’ID scheda
2. Svolgere il lavoro
3. Chiamare g.queue.commit() o g.queue.abort()

scripts/scrape-page.js

async function main() {
  const ctx = g.worker.getContext();
  const { item, tabId, queueName } = ctx;

  // Naviga nella scheda di questo processo di lavoro
  await g.nav.navigateToUrl(item.data.url, { browserTabId: tabId });
  await g.dom.waitForDomStable({ browserTabId: tabId });

  // Estrai i dati
  const title = await g.dom.extractContent("h1", { browserTabId: tabId });

  // Salva il risultato
  g.store.put("results", item.data.url, {
    title: title,
    scrapedAt: new Date().toISOString()
  });

  // Segna come completato
  g.queue.commit(item.id);
}

Non esiste conferma automatica: devi confermare o annullare esplicitamente ogni elemento.

Gestione degli errori

Se lo script annulla un elemento e l’elemento ha superato maxRetries, viene spostato nella coda degli scarti (dead_{queueName}). Puoi ispezionare e riprovare gli elementi scartati:

const deadItems = g.queue.deadLetter("pages");
g.log("Elementi falliti: " + deadItems.length);

// Riprova un elemento specifico
g.queue.retry("pages", deadItems[0].deadLetterId);

Controllare i processi di lavoro

Metodo	Descrizione
g.workers.stop()	Ferma tutti i processi di lavoro
g.workers.stop("pages")	Ferma i processi di lavoro per una coda specifica
g.workers.pause()	Mette in pausa le estrazioni (gli elementi in corso terminano)
g.workers.resume()	Riprende le estrazioni
g.workers.status()	Ottiene lo stato di tutti i pool

Lo stato restituisce un array di istantanee dei pool:

{
  queue: string
  script: string
  concurrency: number
  activeWorkers: number
  processed: number
  failed: number
  remaining: number
  paused: boolean
}

Gestione delle schede

I processi di lavoro creano automaticamente una scheda del browser per ogni processo concorrente. L’opzione layout le dispone così:

• horizontal — schede affiancate
• vertical — schede impilate
• quad — griglia 2x2 (per 4 processi di lavoro)

Imposta clearTabs: true per chiudere tutte le schede esistenti prima dell’avvio.

Limitazione della velocità

L’impostazione throttle controlla la velocità con cui ogni processo di lavoro estrae elementi dalla coda. Un valore di 30 significa che ogni processo attende almeno 2 secondi tra un’estrazione e la successiva. Con concurrency: 3 e throttle: 30, il pool elabora fino a 90 elementi al minuto in totale.

Monitoraggio e operazioni

Ogni esecuzione di elemento da parte di un processo di lavoro appare come attività nel pannello Gestore attività (Task Manager), con durata in tempo reale e pulsanti di arresto.

Per le operazioni quotidiane sulle code, usa la console delle code. Mostra conteggi delle code, elementi in attesa, coda degli scarti, contenuti selezionati, stato dei processi di lavoro, controlli esegui uno/avvia/ferma e collegamenti agli elementi del registro di tracciabilità dei flussi di lavoro quando i dati degli elementi di coda contengono metadati del flusso di lavoro.

Se uno script di processo di lavoro fallisce dopo avere estratto un elemento, Guida annulla l’estrazione quando possibile, così l’elemento di coda non rimane bloccato. Fermare un’attività Esegui uno (Run One) o un pool di processi di lavoro dovrebbe fermare il lavoro futuro e riportare il lavoro di coda in corso a una condizione ispezionabile.

Quando gli elementi di coda sono collegati agli elementi del registro di tracciabilità dei flussi di lavoro, i percorsi di nuovo tentativo ed esecuzione singola della coda verificano prima che l’elemento del registro possa essere preso in carico. Se Guida non riesce a riconciliare l’elemento di coda con il registro, espone un errore rivolto all’operatore invece di creare lavoro di coda destinato a fallire.

Per code appartenenti a un flusso di lavoro, imposta requiresWorkflowEnvelope: true in queues.json o g.workers.start(). Guida rifiuta quindi di elaborare elementi di coda se l’elemento non include metadati del flusso di lavoro con workflowName, runId, itemId e itemKey.

Questo è utile per elaborazioni di raccolta dati di lunga durata in cui lavoro di coda e dati del registro di tracciamento del flusso di lavoro devono restare sincronizzati. Il lavoro estratto non valido viene bloccato e reso ispezionabile come errore invece di passare attraverso nuovi tentativi destinati a fallire.

Per pipeline grandi che attraversano più cartelle di flusso di lavoro, code, pool di processi e fasi del registro, modella l’intento operativo in Flussi di lavoro. La Workflow Control Console può mostrare quale coda o processo richiede attenzione e aprire la Queue Console sulle prove rilevanti.

Revisione manuale

Per code che richiedono giudizio umano, le viste in modalità revisione offrono pulsanti di azione, scorciatoie da tastiera e modalità di revisione multi-scheda. Vedi Code e revisione.

Passi successivi

• Code e revisione — gestisci code, ispeziona contenuti e rivedi elementi
• Registro di tracciabilità dei flussi di lavoro — ispeziona i dati persistenti dei flussi di lavoro e la cronologia di ripristino
• Flussi di lavoro — descrivi operazioni a livello di operatore e passaggi tra code/registro
• Mettere URL in coda ed elaborarli — esempio completo di accodamento, elaborazione e conferma degli elementi di lavoro
• Eseguire processi di lavoro paralleli — avvia più processi di lavoro del browser da uno script
• Riferimento API — g.queue.* — operazioni sulle code
• Riferimento API — g.workers.* — metodi di gestione dei processi di lavoro
• Spazi di lavoro — configurazione dello spazio di lavoro per queues.json