sqs

AWS SQS

Amazon Simple Queue Service（SQS）是一项完全托管的消息队列服务，用于解耦和扩展微服务、分布式系统及无服务器应用。

目录

• 核心概念
• 常见模式
• CLI 参考
• 最佳实践
• 故障排除
• 参考资料

核心概念

队列类型

类型	描述	使用场景
标准队列	至少一次投递，尽力保持顺序	高吞吐量
FIFO 队列	恰好一次投递，严格排序	对顺序敏感的处理

关键设置

设置	描述	默认值
可见性超时	消息接收后被隐藏的时间	30 秒
消息保留时间	消息保留的时长	4 天（最长 14 天）
延迟秒数	消息可用前的延迟时间	0
最大消息大小	消息的最大大小	256 KB

死信队列（DLQ）

用于存放在达到 maxReceiveCount 次尝试后仍处理失败的消息的队列。

常见模式

创建标准队列

AWS CLI：

aws sqs create-queue \
  --queue-name my-queue \
  --attributes '{
    "VisibilityTimeout": "60",
    "MessageRetentionPeriod": "604800",
    "ReceiveMessageWaitTimeSeconds": "20"
  }'

boto3：

import boto3

sqs = boto3.client('sqs')

response = sqs.create_queue(
    QueueName='my-queue',
    Attributes={
        'VisibilityTimeout': '60',
        'MessageRetentionPeriod': '604800',
        'ReceiveMessageWaitTimeSeconds': '20'  # Long polling
    }
)
queue_url = response['QueueUrl']

创建 FIFO 队列

aws sqs create-queue \
  --queue-name my-queue.fifo \
  --attributes '{
    "FifoQueue": "true",
    "ContentBasedDeduplication": "true"
  }'

配置死信队列

# Create DLQ
aws sqs create-queue --queue-name my-queue-dlq

# Get DLQ ARN
DLQ_ARN=$(aws sqs get-queue-attributes \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/my-queue-dlq \
  --attribute-names QueueArn \
  --query 'Attributes.QueueArn' --output text)

# Set redrive policy on main queue
aws sqs set-queue-attributes \
  --queue-url https://sqs.us-east-1.amazonaws.com/123456789012/my-queue \
  --attributes "{
    \"RedrivePolicy\": \"{\\\"deadLetterTargetArn\\\":\\\"${DLQ_ARN}\\\",\\\"maxReceiveCount\\\":\\\"3\\\"}\"
  }"

发送消息

import boto3
import json

sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'

# Send single message
sqs.send_message(
    QueueUrl=queue_url,
    MessageBody=json.dumps({'order_id': '12345', 'action': 'process'}),
    MessageAttributes={
        'MessageType': {
            'DataType': 'String',
            'StringValue': 'Order'
        }
    }
)

# Send to FIFO queue
sqs.send_message(
    QueueUrl='https://sqs.us-east-1.amazonaws.com/123456789012/my-queue.fifo',
    MessageBody=json.dumps({'order_id': '12345'}),
    MessageGroupId='order-12345',
    MessageDeduplicationId='unique-id-12345'
)

# Batch send (up to 10 messages)
sqs.send_message_batch(
    QueueUrl=queue_url,
    Entries=[
        {'Id': '1', 'MessageBody': json.dumps({'id': 1})},
        {'Id': '2', 'MessageBody': json.dumps({'id': 2})},
        {'Id': '3', 'MessageBody': json.dumps({'id': 3})}
    ]
)

接收和处理消息

import boto3
import json

sqs = boto3.client('sqs')
queue_url = 'https://sqs.us-east-1.amazonaws.com/123456789012/my-queue'

while True:
    # Long polling (wait up to 20 seconds)
    response = sqs.receive_message(
        QueueUrl=queue_url,
        MaxNumberOfMessages=10,
        WaitTimeSeconds=20,
        MessageAttributeNames=['All'],
        AttributeNames=['All']
    )

    messages = response.get('Messages', [])

    for message in messages:
        try:
            body = json.loads(message['Body'])
            print(f"Processing: {body}")

            # Process message...

            # Delete on success
            sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=message['ReceiptHandle']
            )
        except Exception as e:
            print(f"Error processing message: {e}")
            # Message will become visible again after visibility timeout

Lambda 集成

# Create event source mapping
aws lambda create-event-source-mapping \
  --function-name my-function \
  --event-source-arn arn:aws:sqs:us-east-1:123456789012:my-queue \
  --batch-size 10 \
  --maximum-batching-window-in-seconds 5

Lambda 处理程序：

def handler(event, context):
    for record in event['Records']:
        body = json.loads(record['body'])
        message_id = record['messageId']

        try:
            process_message(body)
        except Exception as e:
            # Raise to put message back in queue
            raise

    return {'batchItemFailures': []}

CLI 参考

队列管理

命令	描述
aws sqs create-queue	创建队列
aws sqs delete-queue	删除队列
aws sqs list-queues	列出队列
aws sqs get-queue-url	按名称获取队列 URL
aws sqs get-queue-attributes	获取队列设置
aws sqs set-queue-attributes	更新队列设置

消息处理

命令	描述
aws sqs send-message	发送单条消息
aws sqs send-message-batch	最多发送 10 条消息
aws sqs receive-message	接收消息
aws sqs delete-message	删除消息
aws sqs delete-message-batch	最多删除 10 条消息
aws sqs purge-queue	删除所有消息

可见性

命令	描述
aws sqs change-message-visibility	更改超时时间
aws sqs change-message-visibility-batch	批量更改

最佳实践

消息处理

• 使用长轮询（WaitTimeSeconds=20）以减少 API 调用
• 成功处理后及时删除消息
• 配置适当的可见性超时（大于处理时间）
• 为至少一次投递实现幂等消费者

死信队列

• 始终为生产队列配置 DLQ
• 设置适当的 maxReceiveCount（通常为 3-5）
• 使用 CloudWatch 警报监控 DLQ 深度
• 手动或通过自动化处理 DLQ 消息

FIFO 队列

• 使用消息组 ID 对顺序进行分区
• 启用基于内容的重复数据删除或提供重复数据删除 ID
• 吞吐量：不使用批处理时为 300 条消息/秒，使用批处理时为 3000 条消息/秒

安全

• 使用队列策略控制访问
• 使用 SSE-SQS 或 SSE-KMS 启用加密
• 使用 VPC 端点进行私有访问

故障排除

未收到消息

原因：

• 短轮询返回空结果
• 所有消息都在处理中（可见性超时）
• 消息被延迟（DelaySeconds）

调试：

# Check queue attributes
aws sqs get-queue-attributes \
  --queue-url $QUEUE_URL \
  --attribute-names All

# Check approximate message counts
aws sqs get-queue-attributes \
  --queue-url $QUEUE_URL \
  --attribute-names \
    ApproximateNumberOfMessages,\
    ApproximateNumberOfMessagesNotVisible,\
    ApproximateNumberOfMessagesDelayed

消息进入 DLQ

原因：

• 处理错误
• 可见性超时时间过短
• 消费者未删除消息

从 DLQ 重新投递：

# Enable redrive allow policy on source queue
aws sqs set-queue-attributes \
  --queue-url $MAIN_QUEUE_URL \
  --attributes '{"RedriveAllowPolicy": "{\"redrivePermission\":\"allowAll\"}"}'

# Start redrive
aws sqs start-message-move-task \
  --source-arn arn:aws:sqs:us-east-1:123456789012:my-queue-dlq \
  --destination-arn arn:aws:sqs:us-east-1:123456789012:my-queue

重复处理

解决方案：

• 使用 FIFO 队列实现恰好一次投递
• 在消费者中实现幂等性
• 在数据库中跟踪已处理的消息 ID

Lambda 未处理消息

# Check event source mapping
aws lambda list-event-source-mappings \
  --function-name my-function

# Check for errors
aws lambda get-event-source-mapping \
  --uuid <mapping-uuid>

References

• SQS Developer Guide
• SQS API Reference
• SQS CLI Reference
• boto3 SQS