Introdução
Tarefas repetidas para o mesmo pedido devem preservar o resultado de negócio concluído. Você adicionará uma proteção de gravação atômica a um consumidor fornecido, testará pedidos repetidos e independentes e verificará a confirmação bem-sucedida das mensagens.
Conclua primeiro Isole Tarefas com Falha em uma Fila de Mensagens Mortas, Impeça pedidos duplicados com gravações condicionais e Configure e Diagnostique uma Função Lambda. Esta VM independente fornece o consumidor inicial, a função IAM e a tabela de pedidos vazia.
Relação com as certificações
Este laboratório oferece prática nos seguintes tópicos de exame.
- Solutions Architect – Associate (SAA-C03) · Tarefa 2.1: Consumidores idempotentes e proteção de resultados de negócio.
- Developer – Associate (DVA-C02) · Tarefa 1.1: Consumidores idempotentes e proteção de resultados de negócio.
- DevOps Engineer – Professional (DOP-C02) · Tarefa 5.1: Prática dos fundamentos: Consumidores idempotentes e proteção de resultados de negócio.
- Solutions Architect – Professional (SAP-C02) · Tarefa 2.4: Prática dos fundamentos: Consumidores idempotentes e proteção de resultados de negócio.
Conecte uma Fila Vazia ao Consumidor
Nesta etapa, crie uma fila e conecte o consumidor fornecido antes de enviar tarefas de negócio.
Use AWS View ao lado do Terminal para comparar as filas atuais, os resultados do consumidor e os pedidos armazenados. Preserve os dados de referência não relacionados.
cd /home/labex/project
Crie a fila Standard. A substituição de comando, $(...), salva a URL retornada da fila para operações posteriores:
QUEUE_URL=$(aws sqs create-queue --queue-name labex-q05-jobs --attributes VisibilityTimeout=30 --query QueueUrl --output text)
Selecione seu ARN e crie um mapeamento de origem de eventos com tamanho de lote um:
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)
Inspecione a conexão:
aws lambda get-event-source-mapping --uuid "$MAPPING_ID" --query '{Source:EventSourceArn,Function:FunctionArn,Batch:BatchSize,State:State}'
Espere a origem labex-q05-jobs, labex-q05-worker, tamanho de lote 1 e estado Enabled. A função IAM fornecida pode consultar e excluir mensagens apenas nesta fila e gravar apenas na tabela de pedidos, além de registrar logs de execução. AWS View mostra uma fila vazia, nenhum pedido e nenhuma decisão de gravação. Não envie uma tarefa até implantar a proteção na próxima etapa.
Implante uma Proteção Atômica por Chave de Negócio
Nesta etapa, substitua a gravação incondicional do consumidor e comprove que a primeira tarefa protegida é bem-sucedida.
Mensagens distintas da fila podem carregar o mesmo id de negócio. O consumidor confirma uma duplicata correspondente sem gravar o pedido novamente.
Um consumidor idempotente preserva o mesmo resultado de negócio quando chega trabalho repetido. O DynamoDB avalia ConditionExpression='attribute_not_exists(id)' na mesma operação de gravação de PutItem. Isso evita uma condição de corrida entre uma leitura e uma gravação separadas. O id de negócio é a chave; usar o messageId do SQS permitiria que uma nova cópia do mesmo pedido enviada pelo produtor gravasse novamente.
Quando uma condição falha, ReturnValuesOnConditionCheckFailure='ALL_OLD' retorna o item existente. O código captura apenas ConditionalCheckFailedException e compara esse item ao item solicitado. Detalhes idênticos representam uma duplicata já concluída; detalhes conflitantes ou outro erro continuam causando falha, em vez de serem confirmados silenciosamente.
Escreva o consumidor completo abaixo. cat > app.py substitui o arquivo, e o here-document fornece seu conteúdo até o terminador PY. As aspas em 'PY' impedem a expansão pelo shell dentro do código 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
A análise do evento e as verificações de quantidade são o contexto fornecido. A mudança principal é o único put_item condicional, o tratamento restrito de sua falha e o resultado processed/duplicate. Uma duplicata retorna com sucesso para que o consumidor possa confirmar sua cópia SQS depois que o resultado de negócio já tiver sido preservado.
Empacote app.py na raiz do ZIP. O manipulador da função existente é app.handler, portanto o nome do arquivo do módulo importa:
zip -q function.zip app.py
Envie o ZIP binário com fileb://:
aws lambda update-function-code --function-name labex-q05-worker --zip-file fileb://function.zip --query CodeSha256 --output text
O hash retornado identifica o código implantado; o processamento real comprovará seu comportamento. Envie o primeiro pedido:
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"dedup-order","quantity":4}'
Observe AWS View até a fila ficar vazia, o consumidor informar processed: true e o pedido aparecer. Depois, leia-o:
aws dynamodb get-item --table-name labex-q05-orders --key '{"id":{"S":"dedup-order"}}' --consistent-read --query Item
Espere quantidade 4 e total 1100. A decisão de gravação mostra a condição, Accepted, nenhum item anterior e esse item armazenado depois. Uma condição configurada ou um arquivo enviado, por si só, não comprova trabalho bem-sucedido.
Consuma Tarefas Repetidas sem Repetir a Gravação
Nesta etapa, envie duas novas cópias do mesmo pedido de negócio e comprove que outro pedido ainda é concluído de forma independente.
Envie o mesmo conteúdo de negócio duas vezes. Cada resposta de SendMessage tem um novo ID de mensagem SQS, mas a chave do pedido continua sendo 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}'
Envie uma chave de negócio independente:
aws sqs send-message --queue-url "$QUEUE_URL" --message-body '{"id":"other-order","quantity":1}'
Observe AWS View até quatro execuções reais retornarem e a fila ficar vazia. As duas execuções repetidas informam processed: false, duplicate: true. O DynamoDB rejeita as duas gravações condicionais com ConditionalCheckFailedException; cada item antes e depois permanece inalterado. O primeiro pedido e other-order têm, cada um, uma gravação aceita. O consumidor confirma as cópias repetidas em vez de tentar novamente um trabalho já concluído.
Leia os resultados de negócio:
aws dynamodb scan --table-name labex-q05-orders --query Items
Espere exatamente dedup-order/4/1100 e other-order/1/350; a ordem dos itens pode variar. Inspecione os resultados reais do consumidor. Um evento de log pode conter várias linhas; esse pipeline envia a saída JSON nativa para jq -r, divide cada evento em linhas e seleciona apenas as linhas RESULT, deixando os identificadores de recebimento fora dos resultados exibidos:
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 "))'
Verifique a confirmação das mensagens:
aws sqs get-queue-attributes --queue-url "$QUEUE_URL" --attribute-names ApproximateNumberOfMessages ApproximateNumberOfMessagesNotVisible
As duas contagens são zero. Houve quatro entregas reais de mensagens e execuções da função, duas gravações de negócio aceitas e duas rejeições condicionais nativas. Contar apenas os itens da tabela não revelaria sobrescritas incondicionais; use também as decisões reais e os resultados correspondentes do consumidor.
A proteção cobre um item de negócio. Ela não transforma a gravação no DynamoDB e a confirmação no SQS em uma única operação atômica. O registro do pedido é a proteção durável contra duplicatas. Removê-lo permite que a mesma chave crie um item novamente, portanto a retenção e o projeto das chaves de negócio fazem parte da política de um sistema real. Essas repetições sintéticas demonstram trabalho seguro para o mesmo negócio; não estabelecem uma garantia de concorrência em produção nem de exatamente uma vez de ponta a ponta.
Remova a Conexão do Consumidor e os Resultados de Negócio
Nesta etapa, remova a conexão e os recursos que você criou após confirmar o tratamento de duplicatas.
Interrompa seu mapeamento de origem de eventos antes de excluir sua fila de origem:
aws lambda delete-event-source-mapping --uuid "$MAPPING_ID" --query UUID --output text
aws sqs delete-queue --queue-url "$QUEUE_URL"
Remova os dois itens sintéticos de negócio e os logs de execução deste laboratório:
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
Confirme a ausência dos recursos e preserve os dados de referência não relacionados:
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
Os mapeamentos e os pedidos estão vazios, nenhuma URL de fila permanece e o item de referência ainda diz keep unchanged. A função e as estruturas das tabelas fornecidas permanecem no ambiente. Um erro de rede ou de autenticação não comprova a exclusão. AWS View mantém as decisões históricas de gravação enquanto mostra os recursos atuais vazios.
Remova seus arquivos comuns de código e de arquivo compactado:
rm -f app.py function.zip
Execute a verificação da limpeza antes de encerrar o ambiente.
Resumo
Você implantou uma condição atômica por chave de negócio do DynamoDB em um consumidor real de fila. O primeiro pedido foi concluído; duas cópias enviadas separadamente foram consumidas e confirmadas após a rejeição condicional nativa preservar o item original. Uma chave diferente ainda produziu seu próprio resultado. Gravações reais, resultados do consumidor e o estado da fila vazia comprovaram o resultado antes da limpeza dos recursos deste laboratório.
O desafio aplica o isolamento de falhas com tentativas limitadas, e o projeto sem servidor combina depois a recuperação pela DLQ com essa proteção por chave de negócio.
