Введение
Повторные задания одного заказа должны сохранять завершённый бизнес-результат. Вы добавите атомарную защиту записи в предоставленного потребителя, проверите повторные и независимые заказы и подтвердите успешное подтверждение сообщений.
Сначала выполните Изоляцию неудачных заданий в очереди недоставленных сообщений, Предотвращение повторных заказов условными записями и Настройку и диагностику функции Lambda. Эта независимая VM предоставляет начальный обработчик, роль и пустую таблицу заказов.
Связь с сертификацией
Эта лабораторная работа помогает на практике изучить следующие темы экзаменов.
- Solutions Architect – Associate (SAA-C03) · Задача 2.1: Идемпотентные потребители и защита бизнес-результатов.
- Developer – Associate (DVA-C02) · Задача 1.1: Идемпотентные потребители и защита бизнес-результатов.
- DevOps Engineer – Professional (DOP-C02) · Задача 5.1: Базовая практика: Идемпотентные потребители и защита бизнес-результатов.
- Solutions Architect – Professional (SAP-C02) · Задача 2.4: Базовая практика: Идемпотентные потребители и защита бизнес-результатов.
Подключение пустой очереди к обработчику
На этом шаге создайте очередь и подключите предоставленного потребителя до отправки бизнес-заданий.
Используйте AWS View рядом с Terminal, чтобы сравнивать текущие очереди, результаты обработчика и сохранённые заказы. Сохраните несвязанные эталонные данные.
cd /home/labex/project
Создайте очередь Standard. Подстановка команды $(...) сохраняет возвращённый URL очереди для последующих операций:
QUEUE_URL=$(aws sqs create-queue --queue-name labex-q05-jobs --attributes VisibilityTimeout=30 --query QueueUrl --output text)
Выберите её ARN и создайте сопоставление источника событий с размером пакета один:
QUEUE_ARN=$(aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names QueueArn --query Attributes.QueueArn --output text)
MAPPING_ID=$(aws lambda create-event-source-mapping --function-name labex-q05-worker --event-source-arn "$QUEUE_ARN" --batch-size 1 --enabled --query UUID --output text)
Проверьте подключение:
aws lambda get-event-source-mapping --uuid "$MAPPING_ID" --query '{Source:EventSourceArn,Function:FunctionArn,Batch:BatchSize,State:State}'
Ожидайте источник labex-q05-jobs, labex-q05-worker, размер пакета 1 и состояние Enabled. Предоставленная роль может опрашивать и удалять сообщения только этой очереди и записывать только в таблицу заказов, а также журналы выполнения. AWS View показывает пустую очередь, отсутствие заказа и решений записи. Не отправляйте задание до развёртывания защиты на следующем шаге.
Развёртывание атомарной защиты бизнес-ключа
На этом шаге замените безусловную запись обработчика и докажите успех первого защищённого задания.
Разные сообщения очереди могут содержать один и тот же бизнес-идентификатор. Обработчик подтверждает совпадающий дубликат, не записывая заказ снова.
Идемпотентный потребитель сохраняет один и тот же бизнес-результат при поступлении повторной работы. DynamoDB оценивает ConditionExpression='attribute_not_exists(id)' в той же операции записи, что и PutItem. Это предотвращает гонку отдельных чтения и записи. Ключ — бизнес-идентификатор id; использование SQS messageId позволило бы новой копии того же заказа, отправленной производителем, снова выполнить запись.
При невыполненном условии ReturnValuesOnConditionCheckFailure='ALL_OLD' возвращает существующую запись. Код перехватывает только ConditionalCheckFailedException и сравнивает эту запись с запрошенной. Одинаковые данные означают уже завершённый дубликат; конфликтующие данные или другая ошибка по-прежнему приводят к ошибке, а не молчаливому подтверждению.
Запишите полный обработчик ниже. cat > app.py заменяет файл, а here-document предоставляет его содержимое до завершающего PY. Кавычки в 'PY' предотвращают подстановки оболочки внутри кода Python:
cat > app.py <<'PY'
import json
import os
import boto3
from botocore.exceptions import ClientError
def handler(event, context):
print('EVENT '+json.dumps(event,sort_keys=True))
if set(event)!={'Records'} or len(event['Records'])!=1:
raise ValueError('Expected one SQS record')
job=json.loads(event['Records'][0]['body'])
print('JOB '+json.dumps(job,sort_keys=True))
if set(job)!={'id','quantity'} or not isinstance(job['id'],str):
raise ValueError('Use an id and quantity')
quantity=job['quantity']
if isinstance(quantity,bool) or not isinstance(quantity,int) or not 1<=quantity<=10:
raise ValueError('Quantity must be an integer from 1 to 10')
database=boto3.client('dynamodb',endpoint_url='http://127.0.0.1:5000',region_name='us-east-1')
item={'id':{'S':job['id']},'quantity':{'N':str(quantity)},'total_cents':{'N':str(quantity*250+100)}}
try:
database.put_item(TableName=os.environ['TABLE_NAME'],Item=item,
ConditionExpression='attribute_not_exists(id)',
ReturnValuesOnConditionCheckFailure='ALL_OLD')
result={'id':job['id'],'quantity':quantity,'processed':True,'duplicate':False}
except ClientError as error:
if error.response['Error']['Code']!='ConditionalCheckFailedException':
raise
if error.response.get('Item')!=item:
raise ValueError('Order ID already exists with different details') from error
result={'id':job['id'],'quantity':quantity,'processed':False,'duplicate':True}
print('RESULT '+json.dumps(result,sort_keys=True))
return result
PY
Разбор события и проверки количества — предоставленный контекст. Ключевое изменение — единственный условный put_item, узкая обработка его ошибки и результат processed/duplicate. Дубликат возвращается успешно, чтобы потребитель мог подтвердить его копию SQS после сохранения бизнес-результата.
Упакуйте app.py в корне ZIP. Обработчик существующей функции — app.handler, поэтому имя файла модуля имеет значение:
zip -q function.zip app.py
Загрузите двоичный ZIP с fileb://:
aws lambda update-function-code --function-name labex-q05-worker --zip-file fileb://function.zip --query CodeSha256 --output text
Возвращённый хеш определяет развёрнутый код; реальная обработка докажет его поведение. Отправьте первый заказ:
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"dedup-order","quantity":4}'
Наблюдайте AWS View, пока очередь не станет пустой, обработчик не сообщит processed: true и не появится заказ. Затем прочитайте его:
aws dynamodb get-item --table-name labex-q05-orders --key '{"id":{"S":"dedup-order"}}' --consistent-read --query Item
Ожидайте количество 4 и сумму 1100. Решение записи показывает условие, Accepted, отсутствие предыдущей записи и эту сохранённую запись после операции. Одно настроенное условие или загруженный файл не может доказать успешную работу.
Потребление повторных заданий без повторения записи
На этом шаге отправьте две новые копии того же бизнес-заказа и докажите независимое завершение другого заказа.
Отправьте те же бизнес-данные дважды. Каждый ответ SendMessage имеет новый идентификатор сообщения SQS, но ключ заказа остаётся dedup-order:
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"dedup-order","quantity":4}'
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"dedup-order","quantity":4}'
Отправьте независимый бизнес-ключ:
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"other-order","quantity":1}'
Наблюдайте AWS View, пока не вернутся четыре реальных выполнения и очередь не станет пустой. Два повторных выполнения сообщают processed: false, duplicate: true. DynamoDB отклоняет обе условные записи с ConditionalCheckFailedException; каждая запись до и после остаётся неизменной. Первый заказ и other-order имеют по одной принятой записи. Потребитель подтверждает повторные копии, не повторяя уже завершённую работу.
Прочитайте бизнес-результаты:
aws dynamodb scan --table-name labex-q05-orders --query Items
Ожидайте ровно dedup-order/4/1100 и other-order/1/350; порядок записей может отличаться. Проверьте реальные результаты обработчика. Событие журнала может содержать несколько строк; этот конвейер передаёт вывод JSON сервиса в jq -r, разделяет каждое событие на строки и выбирает только строки RESULT, исключая дескрипторы получения из отображаемых результатов:
aws logs filter-log-events --log-group-name /aws/lambda/labex-q05-worker --filter-pattern '"RESULT"' --output json | jq -r '.events[].message | split("\n")[] | select(startswith("RESULT "))'
Проверьте подтверждение:
aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
Оба счётчика равны нулю. Было четыре реальных доставки сообщений и выполнения функции, две принятые бизнес-записи и два отклонения условия самим сервисом. Один подсчёт записей таблицы не выявил бы безусловные перезаписи; используйте также реальные решения и соответствующие результаты потребителя.
Защита охраняет одну бизнес-запись. Она не делает запись DynamoDB и подтверждение SQS одной атомарной операцией. Запись заказа — устойчивая защита от дубликатов. Её удаление позволяет тому же ключу снова создать запись, поэтому хранение и проектирование бизнес-ключей — часть политики реальной системы. Эти синтетические повторы демонстрируют безопасное повторное выполнение одной бизнес-операции; они не устанавливают гарантии при конкурентном выполнении в производственной среде или сквозного выполнения ровно один раз.
Удаление подключения потребителя и бизнес-результатов
На этом шаге удалите подключение и ресурсы упражнения после подтверждения обработки дубликатов.
Остановите своё сопоставление источника событий перед удалением его исходной очереди:
aws lambda delete-event-source-mapping --uuid "$MAPPING_ID" --query UUID --output text
aws sqs delete-queue --queue-url "$QUEUE_URL"
Удалите обе синтетические бизнес-записи и журналы выполнения этой работы:
aws dynamodb delete-item --table-name labex-q05-orders --key '{"id":{"S":"dedup-order"}}'
aws dynamodb delete-item --table-name labex-q05-orders --key '{"id":{"S":"other-order"}}'
aws logs delete-log-group --log-group-name /aws/lambda/labex-q05-worker
Подтвердите отсутствие ресурсов и сохраните несвязанные эталонные данные:
aws lambda list-event-source-mappings --function-name labex-q05-worker --query EventSourceMappings
aws sqs list-queues
aws dynamodb scan --table-name labex-q05-orders --query Items
aws dynamodb scan --table-name labex-q05-reference --query Items
Сопоставления и заказы пусты, URL очередей не остаются, а эталонная запись по-прежнему содержит keep unchanged. Предоставленные функция и структуры таблиц остаются в среде. Ошибка сети или аутентификации не может доказать удаление. AWS View сохраняет исторические решения записи, показывая пустые текущие ресурсы.
Удалите свои обычные файлы кода и архива:
rm -f app.py function.zip
Выполните проверку очистки перед завершением среды.
Резюме
Вы развернули атомарное условие бизнес-ключа DynamoDB в действительном потребителе очереди. Первый заказ завершился; две отдельно отправленные копии были получены и подтверждены после того, как отклонение условия самим сервисом сохранило исходную запись. Другой ключ по-прежнему создал собственный результат. Реальные записи, результаты обработчика и пустое состояние очереди доказали результат перед ограниченной очисткой.
Вызов применяет ограниченную изоляцию ошибок, а бессерверный проект затем сочетает восстановление из DLQ с этой защитой бизнес-ключа.
