Amazon SQS: от азов до production-ready архитектуры

@app.post("/register")
async def register_user(user: UserCreate):
    # Создаём пользователя в БД — 50ms
    db_user = await create_user(user)
 
    # Отправляем email — 800ms (ждём SMTP)
    await send_welcome_email(user.email)
 
    # Синхронизируем с CRM — 400ms (внешний API)
    await sync_to_crm(db_user)
 
    # Трекаем событие — 200ms (внешняя аналитика)
    await track_signup(db_user.id)
 
    return db_user  # Пользователь ждал 1450ms вместо 50ms

import boto3
import json
 
# Создаём клиент SQS
sqs = boto3.client('sqs', region_name='us-east-1')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'
 
# Отправляем сообщение
response = sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({
        'user_id': 12345,
        'email': 'user@example.com',
        'action': 'send_welcome_email'
    }),
    MessageAttributes={
        'Priority': {
            'StringValue': 'high',
            'DataType': 'String'
        }
    }
)
 
print(f"Message sent: {response['MessageId']}")
# Вывод: Message sent: 5fea7756-0ea4-451a-a703-a558b933e274

import boto3
import json
import time
 
sqs = boto3.client('sqs', region_name='us-east-1')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'
 
# Постоянно читаем очередь (polling loop)
while True:
    # Получаем до 10 сообщений (long polling — ждём до 20s)
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20,  # Long polling — экономит запросы
        MessageAttributeNames=['All']
    )
 
    messages = response.get('Messages', [])
 
    if not messages:
        print("Нет сообщений, продолжаем polling...")
        continue
 
    for message in messages:
        try:
            # Парсим тело сообщения
            body = json.loads(message['Body'])
            print(f"Обработка: {body}")
 
            # Имитация обработки (отправка email)
            send_email(body['email'])
 
            # Удаляем сообщение из очереди после успешной обработки
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )
            print(f"Сообщение {message['MessageId']} обработано и удалено")
 
        except Exception as e:
            print(f"Ошибка обработки: {e}")
            # Сообщение НЕ удалено → станет видимым снова через Visibility Timeout
            # После maxReceiveCount попыток → отправится в DLQ
 
def send_email(email):
    """Заглушка отправки email"""
    print(f"Email отправлен на {email}")
    time.sleep(0.5)

import boto3
import json
 
sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/notifications'
 
# Producer: API отправляет уведомление
def send_notification(user_id: int, message: str):
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({
            'user_id': user_id,
            'message': message,
            'timestamp': datetime.utcnow().isoformat()
        })
    )
    # API сразу возвращает 200 OK, не дожидаясь отправки email
 
# Consumer: Lambda обрабатывает уведомления
def lambda_handler(event, context):
    for record in event['Records']:
        body = json.loads(record['body'])
 
        # Идемпотентная проверка (защита от дубликатов)
        if is_notification_sent(body['user_id'], body['timestamp']):
            continue
 
        send_email(body['user_id'], body['message'])
        mark_notification_sent(body['user_id'], body['timestamp'])

import boto3
import json
import hashlib
 
sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/orders.fifo'
 
# Producer: отправляем события заказа
def send_order_event(order_id: int, event_type: str, data: dict):
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({
            'order_id': order_id,
            'event_type': event_type,  # created, paid, shipped
            'data': data
        }),
        # MessageGroupId определяет порядок: все события одного заказа в одной группе
        MessageGroupId=str(order_id),
 
        # MessageDeduplicationId гарантирует exactly-once
        MessageDeduplicationId=hashlib.sha256(
            f"{order_id}:{event_type}:{data.get('timestamp')}".encode()
        ).hexdigest()
    )
 
# Пример использования
send_order_event(12345, 'created', {'amount': 5000, 'timestamp': '2025-12-09T10:00:00Z'})
send_order_event(12345, 'paid', {'payment_id': 'pm_abc', 'timestamp': '2025-12-09T10:01:00Z'})
send_order_event(12345, 'shipped', {'tracking': 'TRACK123', 'timestamp': '2025-12-09T11:00:00Z'})
 
# Consumer обрабатывает события СТРОГО В ПОРЯДКЕ для каждого order_id

# Установка на уровне очереди (default для всех сообщений)
sqs.set_queue_attributes(
    QueueUrl=queue_url,
    Attributes={
        'VisibilityTimeout': '60'  # 60 секунд
    }
)
 
# Продление timeout для конкретного сообщения (если обработка долгая)
sqs.change_message_visibility(
    QueueUrl=queue_url,
    ReceiptHandle=receipt_handle,
    VisibilityTimeout=120  # Ещё 2 минуты
)

import boto3
 
sqs = boto3.client('sqs')
 
# 1. Создаём Dead Letter Queue
dlq_response = sqs.create_queue(
    QueueName='my-queue-dlq',
    Attributes={
        'MessageRetentionPeriod': '1209600'  # 14 дней (максимум)
    }
)
dlq_url = dlq_response['QueueUrl']
dlq_arn = sqs.get_queue_attributes(
    QueueUrl=dlq_url,
    AttributeNames=['QueueArn']
)['Attributes']['QueueArn']
 
# 2. Создаём основную очередь с редирект в DLQ
main_queue_response = sqs.create_queue(
    QueueName='my-queue',
    Attributes={
        'VisibilityTimeout': '60',
        'MessageRetentionPeriod': '345600',  # 4 дня
        # Настройка DLQ: после 3 неудачных попыток → в DLQ
        'RedrivePolicy': json.dumps({
            'deadLetterTargetArn': dlq_arn,
            'maxReceiveCount': '3'
        })
    }
)

# Проверяем количество сообщений в DLQ (алерт для DevOps)
def check_dlq_depth():
    response = sqs.get_queue_attributes(
        QueueUrl=dlq_url,
        AttributeNames=['ApproximateNumberOfMessages']
    )
 
    message_count = int(response['Attributes']['ApproximateNumberOfMessages'])
 
    if message_count > 0:
        print(f"⚠️ ALERT: {message_count} messages in DLQ!")
        # Отправить уведомление в Slack/PagerDuty
        send_alert(f"DLQ has {message_count} failed messages")
 
    return message_count

# После фикса бага — переносим сообщения из DLQ обратно в основную очередь
def redrive_from_dlq():
    while True:
        response = sqs.receive_message(
            QueueUrl=dlq_url,
            MaxNumberOfMessages=10
        )
 
        messages = response.get('Messages', [])
        if not messages:
            break
 
        for message in messages:
            # Отправляем в основную очередь
            sqs.send_message(
                QueueUrl=main_queue_url,
                MessageBody=message['Body']
            )
 
            # Удаляем из DLQ
            sqs.delete_message(
                QueueUrl=dlq_url,
                ReceiptHandle=message['ReceiptHandle']
            )
 
        print(f"Redrived {len(messages)} messages from DLQ")

# ❌ Short polling — дорого и неэффективно
while True:
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10
        # WaitTimeSeconds НЕ указан → short polling
    )
    # Если очереди пустая → миллионы пустых запросов в день
 
# ✅ Long polling — дешево и эффективно
while True:
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20  # Ждём до 20 секунд
    )
    # Меньше запросов → меньше стоимость

sqs.set_queue_attributes(
    QueueUrl=queue_url,
    Attributes={
        'ReceiveMessageWaitTimeSeconds': '20'
    }
)

# Lambda function для обработки SQS сообщений
import json
import boto3
 
def lambda_handler(event, context):
    """
    Lambda автоматически получает batch сообщений из SQS
    event['Records'] — список сообщений (до 10 по умолчанию)
    """
 
    for record in event['Records']:
        try:
            # Парсим тело сообщения
            body = json.loads(record['body'])
            message_id = record['messageId']
 
            print(f"Processing message {message_id}: {body}")
 
            # Ваша бизнес-логика
            process_order(body)
 
            # Lambda АВТОМАТИЧЕСКИ удаляет сообщение из очереди после успешного return
 
        except Exception as e:
            print(f"Error processing message {record['messageId']}: {e}")
            # При ошибке Lambda НЕ удаляет сообщение
            # Оно вернётся в очередь и будет повторно обработано
            raise  # Важно: propagate exception для корректной обработки retry
 
def process_order(order_data):
    """Бизнес-логика обработки заказа"""
    print(f"Order {order_data['order_id']} processed")

resource "aws_lambda_event_source_mapping" "sqs_trigger" {
  event_source_arn = aws_sqs_queue.orders.arn
  function_name    = aws_lambda_function.order_processor.arn
 
  # Batch settings
  batch_size                         = 10      # Обрабатывать до 10 сообщений за раз
  maximum_batching_window_in_seconds = 5       # Ждать 5 секунд для накопления batch
 
  # Scaling
  scaling_config {
    maximum_concurrency = 100  # Максимум 100 одновременных Lambda
  }
 
  # Error handling
  function_response_types = ["ReportBatchItemFailures"]  # Частичные ошибки
}

# Пример 1: Email сервис — 10M сообщений/месяц, short polling
# - 10M send_message: ($0.40 / 1M) × 9M = $3.60 (первый 1M бесплатно)
# - 100M receive_message (10 попыток на сообщение): ($0.40 / 1M) × 99M = $39.60
# - ИТОГО: $43.20/месяц
 
# Пример 2: Тот же сервис, long polling (в 10 раз меньше receive запросов)
# - 10M send_message: $3.60
# - 10M receive_message: ($0.40 / 1M) × 9M = $3.60
# - ИТОГО: $7.20/месяц — ЭКОНОМИЯ $36/месяц (83%)

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["sqs:SendMessage", "sqs:GetQueueUrl"],
      "Resource": "arn:aws:sqs:us-east-1:123456789012:my-queue"
    }
  ]
}

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:ChangeMessageVisibility",
        "sqs:GetQueueAttributes"
      ],
      "Resource": "arn:aws:sqs:us-east-1:123456789012:my-queue"
    }
  ]
}

# Создание очереди с шифрованием через AWS KMS
response = sqs.create_queue(
    QueueName='secure-queue',
    Attributes={
        # SSE с управляемым AWS ключом (бесплатно)
        'SqsManagedSseEnabled': 'true',
 
        # ИЛИ с собственным KMS ключом (контроль доступа)
        # 'KmsMasterKeyId': 'arn:aws:kms:us-east-1:123456789012:key/abc-123',
        # 'KmsDataKeyReusePeriodSeconds': '300'  # 5 минут
    }
)

# Terraform: VPC Endpoint для SQS
resource "aws_vpc_endpoint" "sqs" {
  vpc_id            = aws_vpc.main.id
  service_name      = "com.amazonaws.us-east-1.sqs"
  vpc_endpoint_type = "Interface"
 
  subnet_ids = [
    aws_subnet.private_a.id,
    aws_subnet.private_b.id,
  ]
 
  security_group_ids = [aws_security_group.sqs_endpoint.id]
 
  private_dns_enabled = true
}
 
# Consumer в VPC теперь обращается к SQS через приватную сеть
# Никакого интернет-трафика → безопаснее и дешевле

import boto3
 
cloudwatch = boto3.client('cloudwatch')
 
# 1. ApproximateNumberOfMessagesVisible — глубина очереди
response = cloudwatch.get_metric_statistics(
    Namespace='AWS/SQS',
    MetricName='ApproximateNumberOfMessagesVisible',
    Dimensions=[{'Name': 'QueueName', 'Value': 'my-queue'}],
    StartTime=datetime.utcnow() - timedelta(minutes=10),
    EndTime=datetime.utcnow(),
    Period=300,  # 5 минут
    Statistics=['Average', 'Maximum']
)
 
# 2. ApproximateAgeOfOldestMessage — возраст старейшего сообщения
# Если растёт → consumer не успевают обрабатывать
 
# 3. NumberOfMessagesSent — скорость отправки
 
# 4. NumberOfMessagesDeleted — скорость обработки
 
# 5. ApproximateNumberOfMessagesNotVisible — сообщения в обработке

# Алерт: очередь растёт (consumer не справляются)
resource "aws_cloudwatch_metric_alarm" "queue_depth" {
  alarm_name          = "sqs-queue-depth-high"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 300
  statistic           = "Average"
  threshold           = 1000  # > 1000 сообщений в очереди
 
  dimensions = {
    QueueName = "my-queue"
  }
 
  alarm_actions = [aws_sns_topic.alerts.arn]
}
 
# Алерт: старые сообщения (возможно, consumer упали)
resource "aws_cloudwatch_metric_alarm" "old_messages" {
  alarm_name          = "sqs-old-messages"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "ApproximateAgeOfOldestMessage"
  namespace           = "AWS/SQS"
  period              = 60
  statistic           = "Maximum"
  threshold           = 600  # > 10 минут
 
  dimensions = {
    QueueName = "my-queue"
  }
 
  alarm_actions = [aws_sns_topic.alerts.arn]
}
 
# Алерт: сообщения в DLQ (проблемы с обработкой)
resource "aws_cloudwatch_metric_alarm" "dlq_messages" {
  alarm_name          = "sqs-dlq-has-messages"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 1
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 60
  statistic           = "Sum"
  threshold           = 0  # Любое сообщение в DLQ — алерт
 
  dimensions = {
    QueueName = "my-queue-dlq"
  }
 
  alarm_actions = [aws_sns_topic.critical_alerts.arn]
}

# Метрики для визуализации
metrics_to_track = [
    "ApproximateNumberOfMessagesVisible",  # График глубины очереди
    "ApproximateNumberOfMessagesNotVisible",  # Сообщения в обработке
    "NumberOfMessagesSent",  # Throughput (входящие)
    "NumberOfMessagesDeleted",  # Throughput (обработанные)
    "ApproximateAgeOfOldestMessage",  # Latency обработки
]
 
# Формула: Processing Rate = Deleted / Sent (должно быть ~1.0)
# Если < 1.0 → очередь растёт, нужно добавить consumer

import boto3
import json
 
sqs = boto3.client('sqs')
 
# Три очереди с разными приоритетами
QUEUES = {
    'high': 'https://sqs.us-east-1.amazonaws.com/123456789012/tasks-high',
    'medium': 'https://sqs.us-east-1.amazonaws.com/123456789012/tasks-medium',
    'low': 'https://sqs.us-east-1.amazonaws.com/123456789012/tasks-low'
}
 
# Producer: маршрутизация по приоритету
def send_task(task_data, priority='medium'):
    queue_url = QUEUES.get(priority, QUEUES['medium'])
 
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps(task_data)
    )
 
# Consumer: обработка с приоритетами
def process_tasks():
    """
    Сначала проверяем high, потом medium, потом low.
    Если в high есть сообщения — обрабатываем их, не трогая low.
    """
    while True:
        # Пытаемся получить из high-priority
        messages = try_receive(QUEUES['high'])
        if messages:
            process_batch(messages, 'high')
            continue
 
        # Если high пуста → проверяем medium
        messages = try_receive(QUEUES['medium'])
        if messages:
            process_batch(messages, 'medium')
            continue
 
        # Если medium пуста → обрабатываем low
        messages = try_receive(QUEUES['low'])
        if messages:
            process_batch(messages, 'low')
 
        # Пауза перед следующим циклом
        time.sleep(1)
 
def try_receive(queue_url):
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=1  # Короткий long polling
    )
    return response.get('Messages', [])
 
def process_batch(messages, priority):
    print(f"Processing {len(messages)} messages from {priority} priority queue")
    for message in messages:
        process_message(json.loads(message['Body']))
        sqs.delete_message(
            QueueUrl=QUEUES[priority],
            ReceiptHandle=message['ReceiptHandle']
        )

# Consumer обрабатывает в пропорции 70% high, 20% medium, 10% low
def weighted_polling():
    while True:
        # 7 попыток из 10 — high priority
        for _ in range(7):
            if process_from_queue(QUEUES['high']):
                break
 
        # 2 попытки — medium
        for _ in range(2):
            if process_from_queue(QUEUES['medium']):
                break
 
        # 1 попытка — low
        process_from_queue(QUEUES['low'])

# SNS Topic
resource "aws_sns_topic" "order_created" {
  name = "order-created"
}
 
# SQS Queue для email сервиса
resource "aws_sqs_queue" "email_queue" {
  name = "order-email-queue"
}
 
# Подписка SQS на SNS
resource "aws_sns_topic_subscription" "email_subscription" {
  topic_arn = aws_sns_topic.order_created.arn
  protocol  = "sqs"
  endpoint  = aws_sqs_queue.email_queue.arn
}
 
# SQS Policy: разрешить SNS писать в очередь
resource "aws_sqs_queue_policy" "email_queue_policy" {
  queue_url = aws_sqs_queue.email_queue.id
 
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Principal = {
          Service = "sns.amazonaws.com"
        }
        Action   = "sqs:SendMessage"
        Resource = aws_sqs_queue.email_queue.arn
        Condition = {
          ArnEquals = {
            "aws:SourceArn" = aws_sns_topic.order_created.arn
          }
        }
      }
    ]
  })
}
 
# Аналогично создать analytics_queue и webhook_queue

import boto3
import json
 
sns = boto3.client('sns')
topic_arn = 'arn:aws:sns:us-east-1:123456789012:order-created'
 
def publish_order_created(order_data):
    """
    Публикуем событие в SNS — оно автоматически разошлётся
    во все подписанные SQS очереди
    """
    sns.publish(
        TopicArn=topic_arn,
        Message=json.dumps({
            'order_id': order_data['order_id'],
            'user_id': order_data['user_id'],
            'total': order_data['total'],
            'timestamp': datetime.utcnow().isoformat()
        }),
        MessageAttributes={
            'event_type': {
                'StringValue': 'order.created',
                'DataType': 'String'
            }
        }
    )

# Email Service читает из email_queue
def email_service_consumer():
    while True:
        messages = sqs.receive_message(
            QueueUrl='https://sqs.us-east-1.amazonaws.com/.../order-email-queue',
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20
        ).get('Messages', [])
 
        for message in messages:
            # SNS оборачивает сообщение — нужно распарсить
            sns_message = json.loads(message['Body'])
            order_data = json.loads(sns_message['Message'])
 
            send_order_confirmation_email(order_data)
 
            sqs.delete_message(
                QueueUrl='...',
                ReceiptHandle=message['ReceiptHandle']
            )
 
# Analytics Service читает из analytics_queue (независимо)
# Webhook Service читает из webhook_queue (независимо)

import boto3
import json
 
sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/reminders'
 
# Отправить сообщение с задержкой
def send_reminder(user_id, message, delay_seconds):
    """
    delay_seconds: от 0 до 900 (15 минут) для Standard Queue
    для FIFO очередей задержка НЕ поддерживается
    """
    sqs.send_message(
        QueueUrl=queue_url,
        MessageBody=json.dumps({
            'user_id': user_id,
            'message': message
        }),
        DelaySeconds=delay_seconds  # Максимум 15 минут
    )
 
# Пример: напоминание через 10 минут
send_reminder(12345, "Your cart is waiting!", delay_seconds=600)
 
# Для задержек > 15 минут: используйте EventBridge Scheduler или Step Functions

def process_with_retry(message_body, attempt=1):
    """
    При ошибке — отправляем сообщение обратно в очередь
    с экспоненциальной задержкой: 1s, 2s, 4s, 8s, ...
    """
    try:
        process_task(message_body)
    except Exception as e:
        if attempt < 5:
            delay = min(2 ** attempt, 900)  # Максимум 15 минут
 
            sqs.send_message(
                QueueUrl=queue_url,
                MessageBody=json.dumps({
                    **message_body,
                    'retry_attempt': attempt + 1
                }),
                DelaySeconds=delay
            )
            print(f"Retry scheduled in {delay}s (attempt {attempt + 1})")
        else:
            # После 5 попыток — отправляем в DLQ вручную
            send_to_manual_dlq(message_body, error=str(e))

import boto3
import json
import uuid
 
s3 = boto3.client('s3')
sqs = boto3.client('sqs')
 
BUCKET_NAME = 'my-large-messages-bucket'
QUEUE_URL = 'https://sqs.us-east-1.amazonaws.com/123456789012/large-messages'
 
# Producer: большое сообщение → S3, ссылка → SQS
def send_large_message(large_data):
    """
    large_data — может быть несколько МБ
    """
    # 1. Загружаем данные в S3
    object_key = f"messages/{uuid.uuid4()}.json"
    s3.put_object(
        Bucket=BUCKET_NAME,
        Key=object_key,
        Body=json.dumps(large_data),
        ContentType='application/json'
    )
 
    # 2. Отправляем в SQS только ссылку (< 256 KB)
    sqs.send_message(
        QueueUrl=QUEUE_URL,
        MessageBody=json.dumps({
            's3_bucket': BUCKET_NAME,
            's3_key': object_key,
            'size_bytes': len(json.dumps(large_data))
        })
    )
 
    print(f"Large message stored in S3: s3://{BUCKET_NAME}/{object_key}")
 
# Consumer: читаем ссылку из SQS, загружаем данные из S3
def process_large_messages():
    while True:
        messages = sqs.receive_message(
            QueueUrl=QUEUE_URL,
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20
        ).get('Messages', [])
 
        for message in messages:
            pointer = json.loads(message['Body'])
 
            # Загружаем реальные данные из S3
            response = s3.get_object(
                Bucket=pointer['s3_bucket'],
                Key=pointer['s3_key']
            )
            large_data = json.loads(response['Body'].read())
 
            # Обрабатываем
            process_data(large_data)
 
            # Удаляем сообщение из SQS
            sqs.delete_message(
                QueueUrl=QUEUE_URL,
                ReceiptHandle=message['ReceiptHandle']
            )
 
            # (Опционально) Удаляем файл из S3 после обработки
            s3.delete_object(
                Bucket=pointer['s3_bucket'],
                Key=pointer['s3_key']
            )

resource "aws_s3_bucket_lifecycle_configuration" "cleanup" {
  bucket = aws_s3_bucket.large_messages.id
 
  rule {
    id     = "delete-old-messages"
    status = "Enabled"
 
    # Удалять файлы старше 7 дней (если consumer не обработал)
    expiration {
      days = 7
    }
  }
}