A worker is a background process that consumes tasks or messages and performs processing independently of a request-serving process. It lets a web server accept work quickly while processing continues asynchronously. section 14.2.1 section 16.3.2
A queue-backed worker commonly follows this sequence:
- Poll a message queue for work. section 14.2.5 section 16.3.3
- Read the message, validate its data, and fetch any required records or files. section 16.3.3
- Perform the task, such as taking a website screenshot or applying a sepia filter to an image. section 14.2.5 section 16.3.3
- Store the result and update process state where necessary. section 14.2.1 section 16.3.3
- Acknowledge successful completion by deleting the message from the queue. section 14.2.5 section 16.3.3
A worker should acknowledge a message only after completing its task. If the worker fails before acknowledgement, the message becomes available again after its visibility period and another worker can retry it. Multiple workers can consume the same queue, allowing processing capacity to grow as the queue length increases. section 14.2.5 section 16.3.3
Asynchronous screenshot service
A web server places a URL and random ID in SQS, then returns a future S3 link. A worker downloads the URL, creates a PNG, uploads it to S3, and leaves the identifier available for status or result lookup. section 14.2.1
Listing 16.5 Initializing the Imagery worker (worker/worker.js)
const AWS = require('aws-sdk'); #1 const assert = require('assert-plus'); const Jimp = require('jimp'); const fs = require('fs/promises'); const db = new AWS.DynamoDB({}); #2 const s3 = new AWS.S3({}); const sqs = new AWS.SQS({}); const states = { 'processed': processed }; async function processMessages() { #3 let data = await sqs.receiveMessage({ #4 QueueUrl: process.env.ImageQueue, MaxNumberOfMessages: 1 }).promise(); if (data.Messages && data.Messages.length > 0) { var task = JSON.parse(data.Messages[0].Body); var receiptHandle = data.Messages[0].ReceiptHandle; assert.string(task.imageId, 'imageId'); #5 assert.string(task.desiredState, 'desiredState'); let image = await getImage(task.imageId); #6 if (typeof states[task.desiredState] === 'function') { await states[task.desiredState](image); #7 await sqs.deleteMessage({ #8 QueueUrl: process.env.ImageQueue, ReceiptHandle: receiptHandle }).promise(); } else { throw new Error('unsupported desiredState'); } } } async function run() { while (true) { #9 try { await processMessages(); await new Promise(resolve => setTimeout(resolve, 10000)); #10 } catch (e) { #11 console.log('ERROR', e); } } } run(); #1 Loads the Node.js modules (dependencies) #2 Configures the clients to interact with AWS services #3 This function reads messages from the queue, processes them, and finally deletes the message from the queue. #4 Reads one message from the queue; might return an empty result if there are no messages in the queue #5 Makes sure the message contains all the required properties #6 Gets the process data from the database #7 Triggers the state machine #8 If the message was processed successfully, deletes the message from the queue #9 A loop running endlessly #10 Sleeps for 10 seconds #11 Catches all exceptions, ignores them, and tries again
Shows the Node.js worker setup, AWS endpoints, message polling, task validation, and processing loop. section 16.3.3
