非AWS自研实时JSON消息系统对接AWS SQS与DynamoDB方案咨询

Hey there! Since you're relatively new to AWS, let's start with the most efficient, low-maintenance approach first—serverless architecture using Lambda, which avoids the overhead of managing EC2 instances entirely. Here's how to set it up step by step:

1. 基础配置：SQS队列准备 #

• 首先创建一个SQS队列：根据你的需求选择标准队列（高吞吐量，至少一次交付）或FIFO队列（严格顺序，Exactly-Once交付）。因为你的消息是JSON格式的实时消息，标准队列通常足够，除非你有强顺序要求。
• 确保自研消息系统能正确将JSON消息发送到这个SQS队列（可以先手动发送测试消息验证连通性）。

2. 配置Lambda触发器 #

• 在AWS控制台创建一个Lambda函数：选择适合你熟悉的语言（比如Python，上手简单），使用默认的执行角色（后续我们会完善权限）。
• 给Lambda添加SQS触发器：选择你创建的SQS队列，设置批量大小（比如10条，根据消息大小调整），并配置可见性超时（建议设置为Lambda超时时间的1.5倍，比如Lambda超时设为30秒，可见性超时设为45秒，避免消息被重复处理）。

3. Lambda代码编写（Python示例） #

下面是一个简单的代码片段，用于接收SQS消息、解析JSON并写入DynamoDB：

import boto3
import json

dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Your-DynamoDB-Table-Name')

def lambda_handler(event, context):
    # 遍历批量接收的SQS消息
    for record in event['Records']:
        try:
            # 解析SQS消息体（JSON格式）
            message_body = json.loads(record['body'])
            
            # 写入DynamoDB：这里假设你的消息有唯一ID作为分区键，根据实际表结构调整
            response = table.put_item(
                Item={
                    'messageId': message_body['id'],  # 替换为你的分区键字段
                    'content': message_body,
                    'timestamp': message_body.get('timestamp')
                }
            )
            print(f"Successfully wrote message {message_body['id']} to DynamoDB")
        except Exception as e:
            print(f"Error processing message: {str(e)}")
            # 如果处理失败，Lambda会自动重试（根据SQS的重试策略）
    return {
        'statusCode': 200,
        'body': 'Processed messages successfully'
    }

4. 权限配置（关键！） #

• 给Lambda的执行角色添加以下权限：
  • sqs:ReceiveMessage、sqs:DeleteMessage、sqs:GetQueueAttributes（用于处理SQS消息）
  • dynamodb:PutItem（用于写入DynamoDB）
• 不要硬编码AWS密钥，完全依赖IAM角色即可，这是AWS的最佳实践。

5. 监控与调试 #

• 使用CloudWatch监控Lambda的执行日志、错误率，以及SQS的队列长度（如果队列消息堆积，说明Lambda的并发或批量设置需要调整）。
• 可以先发送几条测试消息到SQS，查看Lambda是否正确处理并写入DynamoDB。

如果你有特殊需求必须用EC2（比如需要自定义运行环境、特殊依赖等），这里是具体实施方法：

1. EC2实例配置 #

• 选择合适的EC2实例类型（比如t2.micro用于测试，根据消息吞吐量调整），使用Amazon Linux 2 AMI（自带AWS CLI和SDK）。
• 配置安全组：允许出站访问SQS和DynamoDB（或者更安全的方式是配置VPC端点，避免公网访问）。
• 给EC2实例分配一个IAM角色，赋予sqs:ReceiveMessage、sqs:DeleteMessage、dynamodb:PutItem权限，同样不要在实例上存储密钥。

2. 编写消息处理脚本（Python示例） #

在EC2上安装boto3（pip install boto3），然后编写一个轮询SQS的脚本：

import boto3
import json
import time

sqs = boto3.client('sqs')
dynamodb = boto3.resource('dynamodb')
table = dynamodb.Table('Your-DynamoDB-Table-Name')
queue_url = 'Your-SQS-Queue-URL'

def process_messages():
    while True:
        # 从SQS获取消息（长轮询减少空请求）
        response = sqs.receive_message(
            QueueUrl=queue_url,
            AttributeNames=['All'],
            MaxNumberOfMessages=10,
            WaitTimeSeconds=20
        )
        
        if 'Messages' in response:
            for message in response['Messages']:
                try:
                    message_body = json.loads(message['Body'])
                    # 写入DynamoDB
                    table.put_item(
                        Item={
                            'messageId': message_body['id'],
                            'content': message_body,
                            'timestamp': message_body.get('timestamp')
                        }
                    )
                    # 处理成功后删除消息，避免重复处理
                    sqs.delete_message(
                        QueueUrl=queue_url,
                        ReceiptHandle=message['ReceiptHandle']
                    )
                    print(f"Processed and deleted message {message['MessageId']}")
                except Exception as e:
                    print(f"Error processing message {message['MessageId']}: {str(e)}")
        time.sleep(1)

if __name__ == "__main__":
    process_messages()

3. 运维与高可用 #

• 将脚本设置为系统服务（比如用systemd），确保EC2重启后自动运行。
• 考虑使用Auto Scaling组：根据SQS队列的ApproximateNumberOfMessages指标自动增减EC2实例，避免消息堆积。
• 用CloudWatch监控EC2的CPU、内存，以及SQS的队列长度，设置告警。

给新手的建议：优先选择Lambda方案，因为它不需要你管理服务器，AWS会处理扩缩容和运维，非常适合快速上手。你可以先搭建最小可用版本，测试通过后再逐步优化（比如添加错误处理、调整批量大小等）。

内容的提问来源于stack exchange，提问作者Kitani Islam