Настройка очередей сообщений (Amazon SQS) для веб-приложения

Настройка Amazon SQS как очереди сообщений

Amazon SQS — управляемая очередь сообщений AWS без необходимости администрировать брокер. Два типа: Standard Queue (высокая пропускная способность, возможны дубликаты) и FIFO Queue (гарантированный порядок, ровно один раз).

Создание очереди через Terraform

# Dead Letter Queue
resource "aws_sqs_queue" "dlq" {
  name                      = "myapp-jobs-dlq"
  message_retention_seconds = 1209600  # 14 дней
}

# Основная очередь
resource "aws_sqs_queue" "jobs" {
  name                       = "myapp-jobs"
  visibility_timeout_seconds = 300      # время обработки сообщения
  message_retention_seconds  = 86400    # 24 часа
  receive_wait_time_seconds  = 20       # long polling

  redrive_policy = jsonencode({
    deadLetterTargetArn = aws_sqs_queue.dlq.arn
    maxReceiveCount     = 3  # 3 неудачи → DLQ
  })
}

# FIFO очередь (для задач, требующих порядка)
resource "aws_sqs_queue" "orders_fifo" {
  name                        = "myapp-orders.fifo"
  fifo_queue                  = true
  content_based_deduplication = true
  deduplication_scope         = "messageGroup"
  fifo_throughput_limit       = "perMessageGroupId"
}

# IAM политика для приложения
resource "aws_iam_policy" "sqs_app" {
  name = "myapp-sqs-access"
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Effect = "Allow"
      Action = [
        "sqs:SendMessage",
        "sqs:ReceiveMessage",
        "sqs:DeleteMessage",
        "sqs:GetQueueAttributes",
      ]
      Resource = [aws_sqs_queue.jobs.arn, aws_sqs_queue.dlq.arn]
    }]
  })
}

PHP: AWS SDK

use Aws\Sqs\SqsClient;

class SqsQueue
{
    private SqsClient $client;
    private string $queueUrl;

    public function __construct()
    {
        $this->client = new SqsClient([
            'version' => 'latest',
            'region'  => config('aws.region', 'eu-west-1'),
        ]);
        $this->queueUrl = config('queue.connections.sqs.queue');
    }

    public function send(string $jobClass, array $payload, int $delaySeconds = 0): string
    {
        $result = $this->client->sendMessage([
            'QueueUrl'     => $this->queueUrl,
            'MessageBody'  => json_encode([
                'job'       => $jobClass,
                'payload'   => $payload,
                'attempts'  => 0,
                'sent_at'   => now()->toIso8601String(),
            ]),
            'DelaySeconds' => $delaySeconds,
            'MessageAttributes' => [
                'JobClass' => [
                    'DataType'    => 'String',
                    'StringValue' => $jobClass,
                ],
            ],
        ]);

        return $result['MessageId'];
    }

    public function poll(int $maxMessages = 10): void
    {
        $result = $this->client->receiveMessage([
            'QueueUrl'            => $this->queueUrl,
            'MaxNumberOfMessages' => $maxMessages,
            'WaitTimeSeconds'     => 20,   // long polling
            'VisibilityTimeout'   => 300,  // 5 минут на обработку
        ]);

        foreach ($result->get('Messages') ?? [] as $message) {
            $this->processMessage($message);
        }
    }

    private function processMessage(array $message): void
    {
        try {
            $body = json_decode($message['Body'], true);
            $job  = app($body['job']);
            $job->handle($body['payload']);

            $this->client->deleteMessage([
                'QueueUrl'      => $this->queueUrl,
                'ReceiptHandle' => $message['ReceiptHandle'],
            ]);
        } catch (\Throwable $e) {
            Log::error('SQS job failed', [
                'job'   => $body['job'] ?? 'unknown',
                'error' => $e->getMessage(),
            ]);
            // Сообщение вернётся в очередь после истечения VisibilityTimeout
        }
    }
}

Laravel Queue с SQS

Laravel поддерживает SQS из коробки:

QUEUE_CONNECTION=sqs
AWS_ACCESS_KEY_ID=AKIA...
AWS_SECRET_ACCESS_KEY=secret
AWS_DEFAULT_REGION=eu-west-1
SQS_QUEUE=https://sqs.eu-west-1.amazonaws.com/123456789/myapp-jobs

// Стандартный Laravel Job
class ProcessOrderJob implements ShouldQueue
{
    use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;

    public int $tries = 3;
    public int $timeout = 120;

    public function __construct(private int $orderId) {}

    public function handle(OrderService $service): void
    {
        $service->process($this->orderId);
    }

    public function failed(\Throwable $e): void
    {
        Log::error('Order processing failed', [
            'order_id' => $this->orderId,
            'error'    => $e->getMessage(),
        ]);
    }
}

// Публикация
ProcessOrderJob::dispatch($order->id);

// С задержкой 5 минут
ProcessOrderJob::dispatch($order->id)->delay(now()->addMinutes(5));

SQS + Lambda (serverless обработка)

# Trigger Lambda из SQS
resource "aws_lambda_event_source_mapping" "sqs_lambda" {
  event_source_arn                   = aws_sqs_queue.jobs.arn
  function_name                      = aws_lambda_function.worker.arn
  batch_size                         = 10
  maximum_batching_window_in_seconds = 5
  function_response_types            = ["ReportBatchItemFailures"]
}

# Lambda handler (Python)
import json

def handler(event, context):
    failed_ids = []

    for record in event['Records']:
        try:
            body = json.loads(record['body'])
            process_job(body)
        except Exception as e:
            print(f"Failed: {record['messageId']}: {e}")
            failed_ids.append({'itemIdentifier': record['messageId']})

    # Partial batch failure — только неудачные вернутся в очередь
    return {'batchItemFailures': [{'itemIdentifier': id} for id in failed_ids]}

Мониторинг

CloudWatch метрики SQS: ApproximateNumberOfMessagesVisible (размер очереди), NumberOfMessagesSent, NumberOfMessagesDeleted, ApproximateAgeOfOldestMessage.

resource "aws_cloudwatch_metric_alarm" "sqs_depth" {
  alarm_name          = "sqs-queue-depth"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "ApproximateNumberOfMessagesVisible"
  namespace           = "AWS/SQS"
  period              = 60
  statistic           = "Maximum"
  threshold           = 1000
  alarm_actions       = [aws_sns_topic.alerts.arn]

  dimensions = { QueueName = aws_sqs_queue.jobs.name }
}

Срок реализации

Laravel Queue + SQS: 1 день (драйвер встроен). Кастомный PHP/Node.js consumer с DLQ и мониторингом: 2–3 дня. SQS + Lambda serverless: 2–3 дня.