LabEx
무료 가입

중복 효과 없이 대기열 작업 처리

AWSBeginner
지금 연습하기

소개

같은 주문에 대한 반복 작업은 완료된 비즈니스 결과를 유지해야 합니다. 제공된 소비자에 원자적인 쓰기 방지 장치를 추가하고 반복 주문과 독립적인 주문을 테스트하며 성공적인 확인 처리를 확인합니다.

배달 못한 편지 대기열로 실패한 작업 격리, 조건부 쓰기로 중복 주문 방지, Lambda 함수 설정 및 진단을 먼저 완료하세요. 이 독립적인 VM에는 시작용 작업자, 역할과 빈 주문 테이블이 제공됩니다.

인증 시험 관련 주제

이 실습은 다음 시험 주제에 대한 실습 경험을 제공합니다.

빈 대기열을 작업자에 연결

이 단계에서는 비즈니스 작업을 전송하기 전에 대기열을 만들고 제공된 소비자를 연결합니다.

Terminal 옆에서 AWS View를 사용하여 현재 대기열, 작업자 결과와 저장된 주문을 비교하세요. 관련 없는 참조 데이터를 유지하세요.

cd /home/labex/project

표준 대기열을 만드세요. 명령 치환 $(...)는 반환된 대기열 URL을 이후 작업을 위해 저장합니다.

QUEUE_URL=$(aws sqs create-queue --queue-name labex-q05-jobs --attributes VisibilityTimeout=30 --query QueueUrl --output text)

ARN을 선택하고 크기가 1인 이벤트 소스 매핑을 만드세요.

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에는 빈 대기열이 표시되고 주문이나 쓰기 결정은 없습니다. 다음 단계에서 방지 장치를 배포할 때까지 작업을 전송하지 마세요.

원자적인 비즈니스 키 방지 장치 배포

이 단계에서는 작업자의 무조건 쓰기를 교체하고 첫 번째 보호된 작업이 성공함을 증명합니다.

메시지 ID와 비즈니스 키

서로 다른 대기열 메시지가 같은 비즈니스 id를 담을 수 있습니다. 작업자는 주문을 다시 쓰지 않고 일치하는 중복을 확인 처리합니다.

멱등적인 소비자는 반복된 작업이 도착해도 같은 비즈니스 결과를 유지합니다. DynamoDB는 PutItem과 같은 쓰기 작업 안에서 ConditionExpression='attribute_not_exists(id)'를 평가합니다. 이는 별도로 읽은 후 쓰면서 발생하는 경쟁 상태를 피합니다. 비즈니스 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

fileb://로 바이너리 ZIP을 업로드하세요.

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}'

대기열이 비고 작업자가 processed: true를 보고하며 주문이 나타날 때까지 AWS View를 관찰하세요. 다음으로 이를 읽으세요.

aws dynamodb get-item --table-name labex-q05-orders --key '{"id":{"S":"dedup-order"}}' --consistent-read --query Item

수량 4와 합계 1100이 나와야 합니다. 쓰기 결정에는 조건, Accepted, 이전 항목이 없다는 점과 이후 저장된 항목이 표시됩니다. 설정된 조건이나 업로드한 파일만으로 성공한 작업을 증명할 수는 없습니다.

쓰기를 반복하지 않고 반복 작업 소비

이 단계에서는 같은 비즈니스 주문의 새 복사본 두 개를 전송하고 다른 주문은 여전히 독립적으로 완료됨을 증명합니다.

같은 비즈니스 페이로드를 두 번 전송하세요. 각 SendMessage 응답에는 새 SQS 메시지 ID가 있지만 주문 키는 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

두 개수 모두 0입니다. 실제 메시지 전달과 함수 실행은 네 번, 허용된 비즈니스 쓰기는 두 번, 네이티브 조건부 거부는 두 번이었습니다. 테이블 항목만 세면 무조건 덮어쓰기를 알 수 없습니다. 실제 결정과 이에 대응하는 소비자 결과도 사용하세요.

방지 장치는 비즈니스 항목 하나를 보호합니다. DynamoDB 쓰기와 SQS 확인 처리를 하나의 원자적인 작업으로 만들지는 않습니다. 주문 레코드가 영구적인 중복 방지 장치입니다. 이를 제거하면 같은 키로 항목을 다시 만들 수 있으므로 보존과 비즈니스 키 설계는 실제 시스템 정책의 일부입니다. 이 테스트 반복은 동일한 비즈니스 작업을 안전하게 처리함을 보여 줍니다. 프로덕션 동시성이나 처음부터 끝까지 정확히 한 번 처리되는 것을 보장하지는 않습니다.

네 번의 실제 실행이 두 주문과 기존 항목을 유지하는 두 조건부 거부를 생성하고 대기열은 빈 AWS View 예시

소비자 연결과 비즈니스 결과 제거

이 단계에서는 중복 처리를 확인한 후 소유한 연결과 리소스를 제거합니다.

소스 대기열을 삭제하기 전에 이벤트 소스 매핑을 중지하세요.

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 복구와 이 비즈니스 키 방지 장치를 결합합니다.

관련 AWS 코스
초보자를 위한 AWS 기초

초보자를 위한 AWS 기초

aws
초보자를 위한 AWS S3

초보자를 위한 AWS S3

aws
초보자를 위한 AWS IAM

초보자를 위한 AWS IAM

aws