利用 AWS SQS 和死信队列 (DLQ) 构建可靠的异步消息处理系统完全指南

# producer.py
import boto3
import json
import uuid

# SQS 클라이언트 생성
sqs = boto3.client('sqs', region_name='ap-northeast-2')

# Terraform에서 생성한 SQS 큐 URL
# 실제 환경에서는 환경 변수나 Parameter Store에서 가져오는 것을 권장합니다.
QUEUE_URL = 'https://sqs.ap-northeast-2.amazonaws.com/123456789012/my-app-production-queue'

def send_email_task(recipient_email, subject, body):
    """이메일 발송 작업을 SQS 큐에 추가합니다."""
    
    message_body = {
        'recipient_email': recipient_email,
        'subject': subject,
        'body': body
    }

    try:
        response = sqs.send_message(
            QueueUrl=QUEUE_URL,
            MessageBody=json.dumps(message_body),
            MessageAttributes={
                'type': {
                    'DataType': 'String',
                    'StringValue': 'email'
                }
            },
            # FIFO 큐 사용 시 필요한 파라미터. 표준 큐에서는 불필요.
            # MessageDeduplicationId=str(uuid.uuid4()),
            # MessageGroupId='email_group'
        )
        print(f"Message sent successfully. MessageId: {response['MessageId']}")
        return response['MessageId']
    except Exception as e:
        print(f"Error sending message: {e}")
        return None

if __name__ == '__main__':
    # 정상적인 이메일 발송 작업 요청
    send_email_task(
        recipient_email='[email protected]',
        subject='Welcome to our service!',
        body='Thank you for signing up.'
    )
    
    # [🚨 테스트용] 고의로 오류를 유발하는 잘못된 형식의 메시지 전송
    # 이 메시지는 Consumer에서 처리 실패 후最终적으로 DLQ로 이동하게 됩니다.
    sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody='{"invalid_json": "this will cause a parsing error"' # Missing closing brace
    )
    print("Sent a malformed message to test DLQ.")

# consumer.py
import boto3
import json
import time

sqs = boto3.client('sqs', region_name='ap-northeast-2')
QUEUE_URL = 'https://sqs.ap-northeast-2.amazonaws.com/123456789012/my-app-production-queue'

def process_email_message(message):
    """실제 이메일 발송 로직을 수행합니다."""
    try:
        # [🚨 중요] 메시지 본문 파싱
        # 여기서 JSON 파싱 에러 등이 발생하면 예외 처리되어 메시지가 삭제되지 않음
        task_data = json.loads(message['Body'])
        
        recipient = task_data['recipient_email']
        subject = task_data['subject']
        
        print(f"Sending email to {recipient} with subject '{subject}'...")
        # 여기에 실제 이메일 발송 로직 (e.g., a call to AWS SES or an SMTP library)
        time.sleep(2) # 이메일 발송에 2초가 걸린다고 가정
        print("Email sent successfully.")

        return True # 처리 성공
    
    except json.JSONDecodeError as e:
        print(f"[ERROR] Failed to parse message body: {message['Body']}. Error: {e}")
        return False # 처리 실패
    except KeyError as e:
        print(f"[ERROR] Missing key in message body: {e}. Body: {message['Body']}")
        return False # 처리 실패
    except Exception as e:
        print(f"An unexpected error occurred while sending email: {e}")
        return False # 처리 실패


def main_loop():
    """메인 워커 루프: SQS 큐를 계속 폴링하며 메시지를 처리합니다."""
    print("Worker started. Waiting for messages...")
    while True:
        try:
            # Long Polling을 사용하여 메시지를 수신합니다.
            response = sqs.receive_message(
                QueueUrl=QUEUE_URL,
                MaxNumberOfMessages=10, # 한 번에 최대 10개의 메시지를 가져옴
                WaitTimeSeconds=20, # Long Polling 대기 시간
                AttributeNames=['All'],
                MessageAttributeNames=['All']
            )

            if 'Messages' in response:
                messages = response['Messages']
                print(f"Received {len(messages)} messages.")
                
                for message in messages:
                    # 메시지 처리 시도
                    is_successful = process_email_message(message)
                    
                    # [🚨 매우 중요] 처리가 성공한 경우에만 큐에서 메시지를 삭제합니다.
                    if is_successful:
                        sqs.delete_message(
                            QueueUrl=QUEUE_URL,
                            ReceiptHandle=message['ReceiptHandle']
                        )
                        print(f"Message {message['MessageId']} deleted.")
            else:
                print("No messages in queue. Waiting...")

        except Exception as e:
            print(f"An error occurred in the main loop: {e}")
            time.sleep(5) # 예기치 못한 에러 발생 시 잠시 대기 후 재시도

if __name__ == '__main__':
    main_loop()