Блог

Я пытаюсь ограничить количество параллельно выполняемых функций. Триггер для моего лямбда-выражения — это сообщение в очереди SQS, как вы можете видеть в моем serverless.yml:

   receiver:
    handler: src/receiver.handler
    timeout: 30
    events:
      - sqs:
          arn: ${queueArn}
          batchSize: 1   
    reservedConcurrency: 1
  

Я использовал «reservedConcurrency: 1», чтобы ограничить количество одновременных выполнений до 1. Я проверил в пользовательском интерфейсе AWS конфигурацию функции ‘receiver’, и для параметра ‘резервный параллелизм’ также установлено значение 1.

Не уверен, чего мне не хватает, поскольку я все еще вижу сотни выполнений, выполняемых как из моих журналов, так и из пользовательского интерфейса AWS SQS.

Интеграция SQS / Lambda не предназначена для работы с зарезервированным параллелизмом ниже 5.

Если вы настраиваете зарезервированный параллелизм для своей функции, установите минимум 5 одновременных выполнений, чтобы уменьшить вероятность ошибок регулирования при вызове вашей функции Lambda.

https://docs.aws.amazon.com/lambda/latest/dg/with-sqs.html

Сообщения в полете — это не только те, которые в настоящее время обрабатываются функцией Lambda. Он также включает сообщения, которые терпят неудачу и возвращаются в очередь из-за недостаточного параллелизма, доступного для функции Lambda — они будут находиться в полете, пока они находятся в процессе попыток, даже если большинство попыток завершатся неудачей. Если вы установите для своего зарезервированного параллелизма значение 0, вы должны обнаружить, что вся обработка останавливается, но некоторое количество сообщений остается в полете, поскольку обработка все еще предпринимается.

Эдуард ссылался на статью medium, в которой объясняется больше решений этой проблемы (https://medium.com/@zaccharles/lambda-concurrency-limits-and-sqs-triggers-dont-mix-well-sometimes-eb23d90122e0).

Но я не нашел ни одного из них особенно полезным для моего варианта использования ограничения не конфигурируемых 5 запросов для сообщений SQS, используемых лямбда-интеграцией.

Мое решение:

   processDatastoreHydrationMessages:
   handler: src/hydrateDatastores/processMessagesHandler.handler
   description: "Query SQS for a message and process it."
   memorySize: 440
   events:
     - schedule: rate(2 minutes)
  

Затем внутри обработчика:

 const sqsClient = new SQS({ region: getRegion() });
const documentClient = new DynamoDB.DocumentClient({ region: getRegion() });
console.log('Querying queue');
 
const response = await sqsClient
.receiveMessage({ QueueUrl: hydrationCommon.hydrationQueueUrl, MaxNumberOfMessages: 1, VisibilityTimeout: 900, WaitTimeSeconds: 0 })
.promise();

const [message] = response?.Messages || [];
console.log('received message');

if (!message || !message.ReceiptHandle) {
  return;
}

// //
// process message here
// //

console.log('Deleting message');

await sqsClient.deleteMessage({ QueueUrl: hydrationCommon.hydrationQueueUrl, ReceiptHandle: message.ReceiptHandle }).promise();

console.log('Finished processing messages');
  

Cron гарантирует, что вы выполняете только один запуск Lambda одновременно. Это своего рода антишаблон, поскольку вы существенно замедляете работу вашей системы. Но это может иметь смысл для некоторых типов построения отчетов или других интенсивных задач обработки данных или ETL.