Amazon SQS #

Amazon Simple Queue Service (SQS) is a fully managed message queue service by AWS — no servers to manage, no clusters to configure. SQS is great for decoupling components in a cloud architecture, letting each component scale independently. The AWS SDK for Go v2 (github.com/aws/aws-sdk-go-v2) is the official way to access SQS from a Go application.

SQS Queue Types #

Standard Queue:
  - Nearly unlimited throughput
  - At-least-once delivery (can duplicate)
  - Message order not guaranteed (best-effort)
  - Good for: most use cases

FIFO Queue (name must end with .fifo):
  - Limited throughput (3000 TPS with batching)
  - Exactly-once delivery (automatic deduplication)
  - Message order guaranteed per Message Group ID
  - Good for: e-commerce orders, financial transactions

SQS Queue Characteristics Comparison #

Feature	Standard Queue	FIFO Queue
Throughput	Nearly unlimited	Max 3,000 TPS (with batching)
Delivery Guarantee	At-least-once (messages can duplicate)	Exactly-once (duplicate-free)
Message Order	Best-effort (order not guaranteed)	Fully guaranteed (First-In-First-Out)
Queue Naming	Free	Must end with `.fifo`

Message Lifecycle & SQS Visibility Timeout #

The SQS message processing cycle relies on the Visibility Timeout mechanism to prevent the same message from being processed by another worker at the same time:

flowchart TD
    P["Producer"] -->|"Send Message"| SQS["Amazon SQS Queue"]
    SQS -->|"ReceiveMessage"| Worker["Consumer (Worker)"]
    
    subgraph Timeout["Visibility Timeout Period (Message Hidden)"]
        Worker -->|"Processing Data"| Status{"Success?"}
    end

    Status -->|"Yes (DeleteMessage)"| Delete["Message Permanently Deleted"]
    Status -->|"No / Timeout Expired"| SQS

Installation #

go get github.com/aws/aws-sdk-go-v2
go get github.com/aws/aws-sdk-go-v2/config
go get github.com/aws/aws-sdk-go-v2/service/sqs

Setting Up the SQS Client #

import (
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/sqs"
    "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

func newSQSClient(ctx context.Context) (*sqs.Client, error) {
    // Load config from the environment, ~/.aws/credentials, or an IAM role
    cfg, err := config.LoadDefaultConfig(ctx,
        config.WithRegion("ap-southeast-1"),
    )
    if err != nil {
        return nil, fmt.Errorf("load config: %w", err)
    }

    return sqs.NewFromConfig(cfg), nil
}

// For development/testing with LocalStack
func newLocalSQSClient(ctx context.Context) (*sqs.Client, error) {
    cfg, err := config.LoadDefaultConfig(ctx,
        config.WithRegion("us-east-1"),
        config.WithCredentialsProvider(
            credentials.NewStaticCredentialsProvider("test", "test", ""),
        ),
    )
    if err != nil {
        return nil, err
    }

    return sqs.NewFromConfig(cfg,
        func(o *sqs.Options) {
            o.BaseEndpoint = aws.String("http://localhost:4566") // LocalStack
        },
    ), nil
}

Queue Operations #

// Get the queue URL (required for all operations)
func getQueueURL(ctx context.Context, client *sqs.Client, name string) (string, error) {
    result, err := client.GetQueueUrl(ctx, &sqs.GetQueueUrlInput{
        QueueName: aws.String(name),
    })
    if err != nil {
        return "", fmt.Errorf("get queue url: %w", err)
    }
    return *result.QueueUrl, nil
}

// Create a new queue
func createQueue(ctx context.Context, client *sqs.Client, name string) (string, error) {
    result, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{
        QueueName: aws.String(name),
        Attributes: map[string]string{
            "VisibilityTimeout":             "30",    // seconds
            "MessageRetentionPeriod":        "86400", // 1 day
            "ReceiveMessageWaitTimeSeconds": "20",    // long polling
            // Dead letter queue
            "RedrivePolicy": `{"deadLetterTargetArn":"arn:aws:sqs:ap-southeast-1:123456789:queue-dlq","maxReceiveCount":"3"}`,
        },
    })
    if err != nil {
        return "", err
    }
    return *result.QueueUrl, nil
}

// Create a FIFO queue
func createFIFOQueue(ctx context.Context, client *sqs.Client, name string) (string, error) {
    result, err := client.CreateQueue(ctx, &sqs.CreateQueueInput{
        QueueName: aws.String(name + ".fifo"), // the name must end with .fifo
        Attributes: map[string]string{
            "FifoQueue":                     "true",
            "ContentBasedDeduplication":     "true", // auto-dedup based on the body
            "VisibilityTimeout":             "30",
            "MessageRetentionPeriod":        "86400",
            "ReceiveMessageWaitTimeSeconds": "20",
        },
    })
    if err != nil {
        return "", err
    }
    return *result.QueueUrl, nil
}

Send — Sending Messages #

// Send one message
func sendMessage(ctx context.Context, client *sqs.Client, queueURL string, body interface{}) error {
    data, err := json.Marshal(body)
    if err != nil {
        return err
    }

    _, err = client.SendMessage(ctx, &sqs.SendMessageInput{
        QueueUrl:    aws.String(queueURL),
        MessageBody: aws.String(string(data)),
        // Delivery delay (0-900 seconds)
        DelaySeconds: 0,
        // Message attributes for metadata
        MessageAttributes: map[string]types.MessageAttributeValue{
            "ContentType": {
                DataType:    aws.String("String"),
                StringValue: aws.String("application/json"),
            },
            "Source": {
                DataType:    aws.String("String"),
                StringValue: aws.String("order-service"),
            },
        },
    })
    return err
}

// Send to a FIFO queue
func sendFIFOMessage(ctx context.Context, client *sqs.Client, queueURL string,
    groupID, deduplicationID string, body interface{}) error {

    data, _ := json.Marshal(body)
    _, err := client.SendMessage(ctx, &sqs.SendMessageInput{
        QueueUrl:               aws.String(queueURL),
        MessageBody:            aws.String(string(data)),
        MessageGroupId:         aws.String(groupID),        // order guaranteed within a group
        MessageDeduplicationId: aws.String(deduplicationID), // prevents duplicates
    })
    return err
}

// Batch send — up to 10 messages at once (more cost-effective)
func sendBatch(ctx context.Context, client *sqs.Client, queueURL string, messages []interface{}) error {
    entries := make([]types.SendMessageBatchRequestEntry, 0, len(messages))
    for i, msg := range messages {
        data, _ := json.Marshal(msg)
        entries = append(entries, types.SendMessageBatchRequestEntry{
            Id:          aws.String(fmt.Sprintf("msg-%d", i)),
            MessageBody: aws.String(string(data)),
        })
    }

    result, err := client.SendMessageBatch(ctx, &sqs.SendMessageBatchInput{
        QueueUrl: aws.String(queueURL),
        Entries:  entries,
    })
    if err != nil {
        return err
    }

    if len(result.Failed) > 0 {
        log.Printf("%d messages failed to send", len(result.Failed))
        for _, f := range result.Failed {
            log.Printf("  Failed ID=%s: %s", *f.Id, *f.Message)
        }
    }
    return nil
}

Receive — Receiving and Processing Messages #

// Receive with long polling
func receiveMessages(ctx context.Context, client *sqs.Client, queueURL string,
    handler func(msg types.Message) error) error {

    for {
        result, err := client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:              aws.String(queueURL),
            MaxNumberOfMessages:   10,  // 1-10 messages per request
            WaitTimeSeconds:       20,  // long polling — wait up to 20 seconds
            VisibilityTimeout:     30,  // hide from other consumers for 30 seconds
            MessageAttributeNames: []string{"All"},
            AttributeNames:        []types.QueueAttributeName{"All"},
        })
        if err != nil {
            if ctx.Err() != nil {
                return nil // context cancelled, not an error
            }
            return fmt.Errorf("receive: %w", err)
        }

        for _, msg := range result.Messages {
            if err := handler(msg); err != nil {
                log.Printf("Error processing message %s: %v", *msg.MessageId, err)
                // The message automatically becomes visible again after the
                // VisibilityTimeout, or extend the time if you need longer
                continue
            }

            // Delete the message after successful processing
            if _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
                QueueUrl:      aws.String(queueURL),
                ReceiptHandle: msg.ReceiptHandle,
            }); err != nil {
                log.Printf("Failed to delete message: %v", err)
            }
        }
    }
}

// Extend the visibility timeout when processing takes longer
func extendVisibility(ctx context.Context, client *sqs.Client, queueURL string,
    receiptHandle *string, newTimeout int32) error {

    _, err := client.ChangeMessageVisibility(ctx, &sqs.ChangeMessageVisibilityInput{
        QueueUrl:          aws.String(queueURL),
        ReceiptHandle:     receiptHandle,
        VisibilityTimeout: newTimeout,
    })
    return err
}

// Batch delete — delete many messages at once
func deleteBatch(ctx context.Context, client *sqs.Client, queueURL string,
    messages []types.Message) error {

    entries := make([]types.DeleteMessageBatchRequestEntry, len(messages))
    for i, msg := range messages {
        entries[i] = types.DeleteMessageBatchRequestEntry{
            Id:            aws.String(fmt.Sprintf("del-%d", i)),
            ReceiptHandle: msg.ReceiptHandle,
        }
    }

    result, err := client.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
        QueueUrl: aws.String(queueURL),
        Entries:  entries,
    })
    if err != nil {
        return err
    }

    if len(result.Failed) > 0 {
        log.Printf("%d messages failed to delete", len(result.Failed))
    }
    return nil
}

Complete Example Program — Job Queue #

package main

import (
    "context"
    "encoding/json"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/sqs"
    "github.com/aws/aws-sdk-go-v2/service/sqs/types"
)

type Job struct {
    ID       string    `json:"id"`
    Type     string    `json:"type"`
    Payload  string    `json:"payload"`
    Priority int       `json:"priority"`
    Created  time.Time `json:"created"`
}

type JobResult struct {
    JobID       string        `json:"job_id"`
    Success     bool          `json:"success"`
    Output      string        `json:"output,omitempty"`
    Error       string        `json:"error,omitempty"`
    Duration    time.Duration `json:"duration"`
    ProcessedAt time.Time     `json:"processed_at"`
}

// JobQueue wrapper for SQS
type JobQueue struct {
    client   *sqs.Client
    queueURL string
}

func NewJobQueue(ctx context.Context, queueURL string) (*JobQueue, error) {
    cfg, err := config.LoadDefaultConfig(ctx,
        config.WithRegion("ap-southeast-1"),
    )
    if err != nil {
        return nil, err
    }

    return &JobQueue{
        client:   sqs.NewFromConfig(cfg),
        queueURL: queueURL,
    }, nil
}

func (q *JobQueue) Enqueue(ctx context.Context, job Job) error {
    data, err := json.Marshal(job)
    if err != nil {
        return err
    }

    _, err = q.client.SendMessage(ctx, &sqs.SendMessageInput{
        QueueUrl:    aws.String(q.queueURL),
        MessageBody: aws.String(string(data)),
        MessageAttributes: map[string]types.MessageAttributeValue{
            "JobType": {
                DataType:    aws.String("String"),
                StringValue: aws.String(job.Type),
            },
            "Priority": {
                DataType:    aws.String("Number"),
                StringValue: aws.String(fmt.Sprintf("%d", job.Priority)),
            },
        },
    })
    if err != nil {
        return fmt.Errorf("enqueue job %s: %w", job.ID, err)
    }

    log.Printf("[QUEUE] Job %s (%s) sent", job.ID, job.Type)
    return nil
}

func (q *JobQueue) Process(ctx context.Context, handler func(Job) (string, error)) {
    log.Println("[WORKER] Started processing jobs...")

    for {
        select {
        case <-ctx.Done():
            log.Println("[WORKER] Stopped")
            return
        default:
        }

        result, err := q.client.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
            QueueUrl:              aws.String(q.queueURL),
            MaxNumberOfMessages:   5,
            WaitTimeSeconds:       20,
            VisibilityTimeout:     60,
            MessageAttributeNames: []string{"All"},
        })
        if err != nil {
            if ctx.Err() != nil {
                return
            }
            log.Printf("[WORKER] Receive error: %v", err)
            time.Sleep(5 * time.Second)
            continue
        }

        for _, msg := range result.Messages {
            var job Job
            if err := json.Unmarshal([]byte(*msg.Body), &job); err != nil {
                log.Printf("[WORKER] Error unmarshal: %v", err)
                q.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
                    QueueUrl:      aws.String(q.queueURL),
                    ReceiptHandle: msg.ReceiptHandle,
                })
                continue
            }

            start := time.Now()
            log.Printf("[WORKER] Processing job %s (%s)", job.ID, job.Type)

            output, err := handler(job)
            duration := time.Since(start)

            if err != nil {
                log.Printf("[WORKER] Job %s FAILED (%v): %v", job.ID, duration, err)
                // Let the message return to the queue (no delete)
                // After maxReceiveCount times, it goes to the DLQ
                continue
            }

            log.Printf("[WORKER] Job %s COMPLETED (%v): %s", job.ID, duration, output)

            // Delete the message after success
            q.client.DeleteMessage(ctx, &sqs.DeleteMessageInput{
                QueueUrl:      aws.String(q.queueURL),
                ReceiptHandle: msg.ReceiptHandle,
            })
        }
    }
}

func processJob(job Job) (string, error) {
    // Simulate various job types
    switch job.Type {
    case "send_email":
        time.Sleep(100 * time.Millisecond)
        return fmt.Sprintf("Email sent for payload: %s", job.Payload), nil

    case "resize_image":
        time.Sleep(500 * time.Millisecond)
        return fmt.Sprintf("Image %s resized successfully", job.Payload), nil

    case "generate_report":
        time.Sleep(2 * time.Second)
        return fmt.Sprintf("Report %s generated successfully", job.Payload), nil

    case "sync_data":
        time.Sleep(300 * time.Millisecond)
        return fmt.Sprintf("Data %s synced successfully", job.Payload), nil

    default:
        return "", fmt.Errorf("unknown job type: %s", job.Type)
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())

    // Capture signals for graceful shutdown
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-sigCh
        log.Println("Received signal, shutting down...")
        cancel()
    }()

    // In production use the real queue URL from AWS
    queueURL := "https://sqs.ap-southeast-1.amazonaws.com/123456789/job-queue"

    // For LocalStack development:
    // queueURL := "http://localhost:4566/000000000000/job-queue"

    queue, err := NewJobQueue(ctx, queueURL)
    if err != nil {
        log.Fatal("Create job queue:", err)
    }

    // Send some jobs
    fmt.Println("=== Sending Jobs ===")
    jobs := []Job{
        {ID: "job-001", Type: "send_email", Payload: "[email protected]", Priority: 1, Created: time.Now()},
        {ID: "job-002", Type: "resize_image", Payload: "photo-123.jpg", Priority: 2, Created: time.Now()},
        {ID: "job-003", Type: "generate_report", Payload: "monthly-Q4", Priority: 3, Created: time.Now()},
        {ID: "job-004", Type: "sync_data", Payload: "users-table", Priority: 1, Created: time.Now()},
        {ID: "job-005", Type: "send_email", Payload: "[email protected]", Priority: 1, Created: time.Now()},
    }

    for _, job := range jobs {
        if err := queue.Enqueue(ctx, job); err != nil {
            log.Printf("Failed to enqueue: %v", err)
        }
    }

    // Process the jobs
    fmt.Println("\n=== Processing Jobs ===")
    queue.Process(ctx, processJob)
}

Summary #

Standard Queues for high throughput; FIFO Queues (name ending in .fifo) for guaranteed ordering.
Long polling (WaitTimeSeconds: 20) reduces cost and latency vs short polling — always use it.
Visibility timeouts hide messages from other consumers while being processed; extend with ChangeMessageVisibility for long-running processes.
Delete messages with DeleteMessage after successful processing — at-least-once delivery.
Dead Letter Queues for messages that fail after maxReceiveCount times — configured via RedrivePolicy.
Batch send (SendMessageBatch) and batch delete (DeleteMessageBatch) reduce API call costs.
Message attributes for metadata without parsing the body — useful for consumer-side routing.
FIFO deduplication: ContentBasedDeduplication (SHA-256 of the body) or an explicit MessageDeduplicationId.
MessageGroupId in FIFO queues to guarantee ordering within a group (e.g. per OrderID).
IAM roles are the best way to authenticate on EC2/ECS/Lambda — no hard-coded credentials.

← Previous: RabbitMQ Next: Google Pub/Sub →