Amazon SQS #

Amazon Simple Queue Service (SQS) is AWS’s managed message queue service — no need to manage broker infrastructure like RabbitMQ or Kafka. SQS provides two queue types: Standard (high throughput, at-least-once delivery, ordering not guaranteed) and FIFO (ordering guaranteed, exactly-once processing, lower throughput). From Dart, SQS is accessed via a REST API with AWS Signature Version 4 authentication — the aws_client package simplifies this by providing a client that handles signing automatically.

Package Setup #

dart pub add aws_client

# pubspec.yaml
dependencies:
  aws_client: ^2.0.0

Standard vs FIFO Queues #

flowchart LR
    subgraph Standard Queue
        P1["Producer"] -->|send| SQ["Standard Queue\n• At-least-once delivery\n• Ordering not guaranteed\n• Unlimited throughput\n• Cheaper"]
        SQ -->|receive| C1["Consumer"]
        SQ -->|receive| C2["Consumer"]
    end

    subgraph FIFO Queue
        P2["Producer"] -->|send| FQ["FIFO Queue\n• Exactly-once delivery\n• Ordering guaranteed per group\n• Max 3.000 TPS\n• More expensive"]
        FQ -->|receive| C3["Consumer"]
    end

Aspect	Standard Queue	FIFO Queue
Ordering	Best-effort (not guaranteed)	Strict (FIFO per group)
Delivery	At-least-once (may duplicate)	Exactly-once
Throughput	Unlimited	Max 3.000 TPS (with batching)
Queue name	Free	Must end with `.fifo`
Best for	Email notifications, logs, fan-out	Financial transactions, order processing

AWS Configuration #

import 'package:aws_client/sqs_2012_11_05.dart';
import 'package:aws_client/src/credentials/credentials.dart';

// Method 1: Explicit credentials (for development)
SQS buatKlienSQS() {
  return SQS(
    region: 'ap-southeast-1',  // AWS region, e.g. Asia Pacific (Singapore)
    credentials: AwsClientCredentials(
      accessKey: '<YOUR_ACCESS_KEY>',
      secretKey: '<YOUR_SECRET_KEY>',
    ),
  );
}

// Method 2: Environment variables (recommended for production)
// Set: AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, AWS_REGION
// aws_client automatically reads from the environment
SQS buatKlienSQSDariEnv() {
  return SQS(region: 'ap-southeast-1');
  // credentials read from: env vars → ~/.aws/credentials → EC2 instance profile
}

// Method 3: For AWS Lambda / ECS — use an IAM Role (no explicit credentials)
// Assign an IAM role with SQS permissions to the Lambda/ECS task
// aws_client will use the instance metadata service automatically

Queue Management #

import 'package:aws_client/sqs_2012_11_05.dart';

Future<void> kelolaQueue(SQS sqs) async {
  // Create a Standard Queue
  final standardQueue = await sqs.createQueue(
    queueName: 'order-events',
    attributes: {
      QueueAttributeName.visibilityTimeout: '30',   // 30 seconds
      QueueAttributeName.messageRetentionPeriod: '86400',  // 1 day
      QueueAttributeName.receiveMessageWaitTimeSeconds: '20',  // long polling
      QueueAttributeName.redrivePolicy: jsonEncode({
        'maxReceiveCount': '3',          // try 3 times before going to the DLQ
        'deadLetterTargetArn': 'arn:aws:sqs:ap-southeast-1:123456:order-events-dlq',
      }),
    },
  );
  print('Queue URL: ${standardQueue.queueUrl}');

  // Create a FIFO Queue — the name must end with .fifo
  final fifoQueue = await sqs.createQueue(
    queueName: 'payment-events.fifo',
    attributes: {
      QueueAttributeName.fifoQueue: 'true',
      QueueAttributeName.contentBasedDeduplication: 'true',  // auto-dedup via hash
      QueueAttributeName.visibilityTimeout: '60',
    },
  );

  // List all queues
  final queues = await sqs.listQueues();
  for (final url in queues.queueUrls ?? []) {
    print('Queue: $url');
  }

  // Get the queue URL by name
  final urlResult = await sqs.getQueueUrl(queueName: 'order-events');
  final queueUrl = urlResult.queueUrl!;

  // Queue attribute info
  final attrs = await sqs.getQueueAttributes(
    queueUrl: queueUrl,
    attributeNames: [
      QueueAttributeName.approximateNumberOfMessages,
      QueueAttributeName.approximateNumberOfMessagesNotVisible,
    ],
  );
  print('Available messages: ${attrs.attributes?[QueueAttributeName.approximateNumberOfMessages]}');
}

Sending Messages #

import 'package:aws_client/sqs_2012_11_05.dart';
import 'dart:convert';

class SQSProducer {
  final SQS _sqs;
  final String _queueUrl;

  SQSProducer({required SQS sqs, required String queueUrl})
      : _sqs = sqs,
        _queueUrl = queueUrl;

  // Send a single message to a Standard Queue
  Future<String> kirim(Map<String, dynamic> data) async {
    final result = await _sqs.sendMessage(
      queueUrl: _queueUrl,
      messageBody: jsonEncode(data),
      messageAttributes: {
        'tipeEvent': MessageAttributeValue(
          dataType: 'String',
          stringValue: data['tipe'] as String?,
        ),
        'versi': MessageAttributeValue(
          dataType: 'String',
          stringValue: '1.0',
        ),
      },
      delaySeconds: 0,  // delivery delay (0-900 seconds)
    );

    print('Message sent: ${result.messageId}');
    return result.messageId!;
  }

  // Send to a FIFO Queue — needs MessageGroupId and MessageDeduplicationId
  Future<String> kirimFIFO({
    required Map<String, dynamic> data,
    required String groupId,        // ordering guaranteed per same groupId
    String? deduplicationId,        // unique ID to prevent duplicates (null = auto from content)
  }) async {
    final result = await _sqs.sendMessage(
      queueUrl: _queueUrl,
      messageBody: jsonEncode(data),
      messageGroupId: groupId,
      messageDeduplicationId: deduplicationId ??
          _hashContent(jsonEncode(data)),  // if contentBasedDeduplication is on, can be null
    );

    return result.messageId!;
  }

  // Batch send — send up to 10 messages at once (more cost-efficient)
  Future<void> kirimBatch(List<Map<String, dynamic>> pesanList) async {
    // SQS batch max 10 messages
    for (int i = 0; i < pesanList.length; i += 10) {
      final batch = pesanList.sublist(i, (i + 10).clamp(0, pesanList.length));

      final entries = batch.asMap().entries.map((e) {
        return SendMessageBatchRequestEntry(
          id: 'msg-${e.key}',          // unique ID within the batch
          messageBody: jsonEncode(e.value),
        );
      }).toList();

      final result = await _sqs.sendMessageBatch(
        queueUrl: _queueUrl,
        entries: entries,
      );

      if (result.failed != null && result.failed!.isNotEmpty) {
        for (final gagal in result.failed!) {
          print('Failed to send ${gagal.id}: ${gagal.message}');
        }
      }

      print('Batch sent: ${result.successful?.length ?? 0}/${batch.length}');
    }
  }

  String _hashContent(String content) {
    // Simple hash implementation (use the crypto package for production)
    return content.hashCode.toRadixString(16);
  }
}

Receiving and Processing Messages #

import 'package:aws_client/sqs_2012_11_05.dart';
import 'dart:convert';

class SQSConsumer {
  final SQS _sqs;
  final String _queueUrl;
  bool _jalan = true;

  SQSConsumer({required SQS sqs, required String queueUrl})
      : _sqs = sqs,
        _queueUrl = queueUrl;

  Future<void> mulai(Future<void> Function(Map<String, dynamic>) handler) async {
    print('SQS consumer started');

    while (_jalan) {
      try {
        // Long polling — wait up to 20 seconds if the queue is empty
        // More efficient than short polling which keeps sending requests
        final result = await _sqs.receiveMessage(
          queueUrl: _queueUrl,
          maxNumberOfMessages: 10,        // take up to 10 messages at once
          waitTimeSeconds: 20,            // long polling — saves cost and CPU
          visibilityTimeout: 60,          // hide from other consumers for 60 seconds
          messageAttributeNames: ['All'], // fetch all message attributes
        );

        final pesan = result.messages ?? [];
        if (pesan.isEmpty) continue;

        // Process all received messages
        await Future.wait(
          pesan.map((msg) => _prosesSatuPesan(msg, handler)),
        );

      } catch (e) {
        print('Error while polling: $e');
        await Future.delayed(Duration(seconds: 5)); // backoff on error
      }
    }
  }

  Future<void> _prosesSatuPesan(
    Message msg,
    Future<void> Function(Map<String, dynamic>) handler,
  ) async {
    try {
      final data = jsonDecode(msg.body!) as Map<String, dynamic>;

      print('Processing message: ${msg.messageId}');
      print('Receive count: ${msg.attributes?['ApproximateReceiveCount']}');

      await handler(data);

      // Delete the message after successful processing — like an ACK
      await _sqs.deleteMessage(
        queueUrl: _queueUrl,
        receiptHandle: msg.receiptHandle!,  // unique handle from the receive
      );

      print('Message ${msg.messageId} processed and deleted successfully');

    } catch (e) {
      print('Failed to process message ${msg.messageId}: $e');
      // Don't delete — the visibility timeout will expire and the message becomes visible again
      // After maxReceiveCount failures, the message automatically goes to the DLQ
    }
  }

  void hentikan() => _jalan = false;
}

// Usage
Future<void> main() async {
  final sqs = SQS(region: 'ap-southeast-1');
  final urlResult = await sqs.getQueueUrl(queueName: 'order-events');

  final consumer = SQSConsumer(
    sqs: sqs,
    queueUrl: urlResult.queueUrl!,
  );

  await consumer.mulai((data) async {
    print('Processing: ${data['tipe']}');
    await prosesBisnisLogic(data);
  });
}

Visibility Timeout — How SQS Works #

Understanding the visibility timeout is crucial to avoid losing messages or processing them twice:

// Scenario:
// 1. Consumer A receives a message → the message is "invisible" during the visibilityTimeout (30 seconds)
// 2. Consumer A crashes before deleting the message
// 3. After 30 seconds, the message becomes "visible" again
// 4. Consumer B receives the same message → at-least-once delivery!

// For messages that need longer processing time:
// Extend the visibility timeout before it expires
Future<void> prosesLama(SQS sqs, String queueUrl, Message msg) async {
  // Start a timer to extend every 25 seconds (before the 30 seconds run out)
  final timer = Timer.periodic(Duration(seconds: 25), (_) async {
    await sqs.changeMessageVisibility(
      queueUrl: queueUrl,
      receiptHandle: msg.receiptHandle!,
      visibilityTimeout: 30,  // extend by another 30 seconds
    );
    print('Visibility timeout extended for ${msg.messageId}');
  });

  try {
    await operasiYangLama();  // a process needing more than 30 seconds
    await sqs.deleteMessage(queueUrl: queueUrl, receiptHandle: msg.receiptHandle!);
  } finally {
    timer.cancel();
  }
}

Dead Letter Queues (DLQ) #

import 'package:aws_client/sqs_2012_11_05.dart';

Future<void> setupDLQ(SQS sqs) async {
  // 1. Create the DLQ first
  final dlqResult = await sqs.createQueue(
    queueName: 'order-events-dlq',
    attributes: {
      QueueAttributeName.messageRetentionPeriod: '1209600',  // keep for 14 days
    },
  );
  final dlqUrl = dlqResult.queueUrl!;

  // Get the DLQ ARN
  final dlqAttrs = await sqs.getQueueAttributes(
    queueUrl: dlqUrl,
    attributeNames: [QueueAttributeName.queueArn],
  );
  final dlqArn = dlqAttrs.attributes![QueueAttributeName.queueArn]!;

  // 2. Create the main queue with a redrive policy to the DLQ
  await sqs.createQueue(
    queueName: 'order-events',
    attributes: {
      QueueAttributeName.redrivePolicy: jsonEncode({
        'maxReceiveCount': '3',        // try 3 times, then go to the DLQ
        'deadLetterTargetArn': dlqArn,
      }),
    },
  );

  // 3. Monitor the DLQ — alert if there are messages here
  Future<void> monitorDLQ() async {
    while (true) {
      final attrs = await sqs.getQueueAttributes(
        queueUrl: dlqUrl,
        attributeNames: [QueueAttributeName.approximateNumberOfMessages],
      );
      final jumlah = int.tryParse(
        attrs.attributes?[QueueAttributeName.approximateNumberOfMessages] ?? '0',
      ) ?? 0;

      if (jumlah > 0) {
        print('⚠️ There are $jumlah failed messages in the DLQ!');
        // Send an alert to the team via Slack/PagerDuty
      }

      await Future.delayed(Duration(minutes: 5));
    }
  }
}

SQS is often used together with SNS for the fan-out pattern — one event sent to many queues:

// SNS Topic → several SQS Queues
// One SNS message is distributed to all subscribers in parallel

// Setup via the AWS Console or CloudFormation:
// SNS Topic "order-events"
//   → Subscribe SQS "order-processor"      (process orders)
//   → Subscribe SQS "inventory-updater"    (update stock)
//   → Subscribe SQS "notification-sender"  (send notifications)
//   → Subscribe SQS "analytics-collector"  (record for analytics)

// Each SQS queue processes independently
// If one service is down, other services aren't affected

// Messages from SNS to SQS arrive in an envelope format:
Future<void> prosesPesanDariSNS(Map<String, dynamic> sqsMessage) async {
  // SNS wraps messages in a JSON envelope
  if (sqsMessage.containsKey('Type') && sqsMessage['Type'] == 'Notification') {
    // This is a message from SNS
    final pesanAsli = jsonDecode(sqsMessage['Message'] as String) as Map<String, dynamic>;
    final topicArn = sqsMessage['TopicArn'] as String;
    print('Event from SNS topic: $topicArn');
    await prosesEventAsli(pesanAsli);
  } else {
    // A message sent directly to SQS
    await prosesEventAsli(sqsMessage);
  }
}

Batch Delete — Deleting Many Messages at Once #

// After processing a batch, delete them all at once (saves API costs)
Future<void> prosesDanHapusBatch(
  SQS sqs,
  String queueUrl,
  List<Message> pesan,
  Future<void> Function(Map<String, dynamic>) handler,
) async {
  final berhasil = <String>[];  // receiptHandles that succeeded

  await Future.wait(
    pesan.map((msg) async {
      try {
        final data = jsonDecode(msg.body!) as Map<String, dynamic>;
        await handler(data);
        berhasil.add(msg.receiptHandle!);
      } catch (e) {
        print('Failed to process ${msg.messageId}: $e');
        // Not added to berhasil → not deleted → returns to the queue
      }
    }),
  );

  if (berhasil.isEmpty) return;

  // Batch delete all successful ones — max 10 per request
  final entries = berhasil.asMap().entries.map((e) =>
    DeleteMessageBatchRequestEntry(
      id: 'del-${e.key}',
      receiptHandle: e.value,
    )
  ).toList();

  await sqs.deleteMessageBatch(
    queueUrl: queueUrl,
    entries: entries,
  );

  print('Batch delete: ${berhasil.length}/${pesan.length} messages deleted');
}

SQS Anti-Patterns #

Repeated Short Polling #

// ANTI-PATTERN: short polling — wastes cost and CPU
while (true) {
  final result = await sqs.receiveMessage(
    queueUrl: queueUrl,
    maxNumberOfMessages: 1,
    waitTimeSeconds: 0,  // ✗ short polling — returns immediately even if the queue is empty
  );
  // If the queue is empty, it still sends a request and you still get billed!
  await Future.delayed(Duration(seconds: 1)); // very fast loop
}

// CORRECT: long polling — efficient and cost-effective
while (true) {
  final result = await sqs.receiveMessage(
    queueUrl: queueUrl,
    maxNumberOfMessages: 10,
    waitTimeSeconds: 20,  // ✓ wait up to 20 seconds — saves API call costs
  );
}

Not Deleting Messages After Processing #

// ANTI-PATTERN: forgetting to delete messages after successful processing
final result = await sqs.receiveMessage(queueUrl: queueUrl);
for (final msg in result.messages ?? []) {
  await prosesPesan(msg);
  // ✗ no deleteMessage call → the message becomes visible again after the timeout
  // The message will be processed repeatedly!
}

// CORRECT: always delete after success
for (final msg in result.messages ?? []) {
  await prosesPesan(msg);
  await sqs.deleteMessage(
    queueUrl: queueUrl,
    receiptHandle: msg.receiptHandle!,
  );  // ✓
}

Summary #

Standard vs FIFO — use Standard for high throughput without strict ordering needs; FIFO for transactions requiring ordering and exactly-once (names must end in .fifo).
Long polling (waitTimeSeconds: 20) is mandatory — short polling wastes cost and resources. SQS bills per request, not per message.
Batch operations (sendMessageBatch, deleteMessageBatch) save up to 90% in API costs — send and delete up to 10 messages per request.
Visibility timeouts hide messages from other consumers while being processed — set it longer than the maximum processing duration, or extend dynamically with changeMessageVisibility.
Dead Letter Queues (DLQ) are configured via redrivePolicy.maxReceiveCount — after N failures, messages automatically move to the DLQ. Monitor the DLQ to detect problems.
Always delete messages with deleteMessage after successful processing — otherwise messages get reprocessed after the visibility timeout expires.
IAM Roles are safer than explicit credentials for Lambda, ECS, and EC2 — no secrets to store, automatically rotated by AWS.
SNS + SQS fan-out for distributing events to many services — one SNS publish, many SQS queues receiving independently.
FIFO message groups — messages with the same messageGroupId are processed in order (FIFO), but different groups can be processed in parallel.
contentBasedDeduplication on FIFO queues removes the need for explicit messageDeduplicationId — SQS automatically deduplicates based on content hashes for 5 minutes.

← Previous: RabbitMQ Next: Google Pub/Sub →