sqs

用途

AWS SQS（Amazon Simple Queue Service）是 AWS 提供的全托管消息队列服务，用于解耦和扩展微服务、分布式系统以及无服务器应用。该技能文档面向需要在 AWS 上创建、配置和使用 SQS 队列的开发者，提供从核心概念到具体操作命令的参考。

主要能力

文档涵盖以下内容：

队列类型：标准队列（至少一次投递、尽力排序，适合高吞吐）与 FIFO 队列（精确一次、严格排序，适合顺序敏感处理）。
关键设置：可见性超时（默认 30 秒）、消息保留期（默认 4 天，最长 14 天）、延迟发送（默认 0）、最大消息大小（256 KB）。
死信队列（DLQ）：用于存放超过 maxReceiveCount 次处理失败的消息。
常见模式：创建标准队列、创建 FIFO 队列、配置死信队列、发送消息（单条、FIFO、批量最多 10 条）、接收与处理消息、Lambda 集成。
CLI 参考：队列管理（create-queue、delete-queue、list-queues、get-queue-url、get-queue-attributes、set-queue-attributes）、消息操作（send-message、send-message-batch、receive-message、delete-message、delete-message-batch、purge-queue）、可见性操作（change-message-visibility、change-message-visibility-batch）。
最佳实践：消息处理（长轮询、及时删除、合理可见性超时、幂等消费者）、死信队列（生产队列必配、maxReceiveCount 通常 3-5、CloudWatch 监控、手动或自动处理）、FIFO 队列（消息组 ID 分区、内容去重或去重 ID、吞吐限制）、安全（队列策略、SSE-SQS 或 SSE-KMS 加密、VPC 端点）。
故障排查：消息收不到、消息进入 DLQ、重复处理、Lambda 不处理等场景的原因与调试命令。

使用方式

文档通过 AWS CLI 和 boto3（Python）示例展示操作方式。例如创建标准队列：

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\\\"}\"
  }"

发送消息（boto3）：

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})}
    ]
)

接收与处理消息（boto3）：

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 handler：

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': []}

适用场景

微服务、分布式系统与无服务器应用之间的异步解耦。
需要削峰填谷、缓冲突发流量的场景。
需要可靠投递（至少一次）或严格顺序处理（FIFO）的消息处理。
与 AWS Lambda 集成实现事件驱动处理。
需要死信队列处理失败消息的生产环境。

限制

标准队列为至少一次投递、尽力排序；FIFO 队列为精确一次、严格排序。
FIFO 队列吞吐：无批处理约 300 条/秒，批处理约 3000 条/秒。
消息最大 256 KB。
消息保留期默认 4 天，最长 14 天。
可见性超时默认 30 秒。
批量发送与删除最多 10 条消息。
文档未提供收费、价格或平台支持等具体信息。

技能介绍

用途

主要能力

使用方式

适用场景

限制