如何集成Amazon SQS与Amazon RDS？实现SQS数据转存至RDS MySQL表

1. 前置准备 #

• 确保Amazon RDS MySQL实例处于运行状态，且已配置好访问权限（比如允许Lambda所在VPC的私有子网访问，或按需开放公网访问）
• 已创建目标Amazon SQS队列（标准队列适合高吞吐量，FIFO队列适合需要严格顺序或避免重复消费的场景）
• 配置IAM角色，赋予以下核心权限：
  • 针对SQS队列：sqs:ReceiveMessage、sqs:DeleteMessage、sqs:GetQueueAttributes
  • 针对RDS实例：rds-db:connect
  • 若Lambda部署在VPC内，需额外添加VPC网络权限（如ec2:CreateNetworkInterface、ec2:DescribeNetworkInterfaces）

2. 创建RDS MySQL数据表 #

根据SQS消息的结构设计表结构，假设SQS消息为JSON格式（包含id、content、created_at字段），执行以下SQL：

CREATE TABLE IF NOT EXISTS sqs_messages (
    id VARCHAR(64) PRIMARY KEY COMMENT 'SQS消息唯一ID',
    content TEXT NOT NULL COMMENT '消息原始内容',
    created_at DATETIME NOT NULL COMMENT '消息生成时间',
    inserted_at DATETIME DEFAULT CURRENT_TIMESTAMP COMMENT '写入RDS时间'
);

提示：如果消息可能重复，可通过ON DUPLICATE KEY UPDATE做幂等处理；若消息结构不同，直接调整表字段即可。

3. 用Lambda实现消息消费与RDS插入 #

Lambda是serverless场景下的最优选择，无需管理服务器，可自动触发SQS消息消费。

3.1 配置Lambda运行环境 #

• 选择Python 3.x作为运行时（也可选用Node.js、Java等，以下以Python为例）
• 若RDS在VPC内，将Lambda配置到同一VPC的私有子网，确保网络连通性

3.2 编写Lambda函数代码 #

将RDS配置通过Lambda环境变量传入（避免硬编码），代码如下：

import json
import mysql.connector
import os
from datetime import datetime

def lambda_handler(event, context):
    # 从环境变量读取RDS配置
    db_config = {
        'user': os.environ['DB_USER'],
        'password': os.environ['DB_PASSWORD'],
        'host': os.environ['DB_HOST'],
        'database': os.environ['DB_NAME'],
        'port': int(os.environ['DB_PORT'])
    }

    try:
        # 建立MySQL连接
        conn = mysql.connector.connect(**db_config)
        cursor = conn.cursor()

        # 批量插入SQL（支持幂等更新）
        insert_sql = """
            INSERT INTO sqs_messages (id, content, created_at)
            VALUES (%s, %s, %s)
            ON DUPLICATE KEY UPDATE content = VALUES(content), created_at = VALUES(created_at)
        """

        # 批量处理SQS消息
        batch_records = []
        for msg in event['Records']:
            msg_body = json.loads(msg['body'])
            batch_records.append(
                (msg['messageId'], json.dumps(msg_body), datetime.fromisoformat(msg_body['created_at']))
            )

        # 执行批量插入
        if batch_records:
            cursor.executemany(insert_sql, batch_records)
            conn.commit()
            print(f"成功处理 {cursor.rowcount} 条记录")

        # 关闭连接
        cursor.close()
        conn.close()

    except Exception as e:
        print(f"处理失败: {str(e)}")
        # 抛出异常触发Lambda重试（需配合SQS可见性超时配置）
        raise e

    return {
        'statusCode': 200,
        'body': f"完成 {len(event['Records'])} 条消息处理"
    }

注意：若使用mysql-connector-python依赖，需打包成Lambda层上传，或使用Lambda自带的依赖版本。

3.3 配置Lambda触发SQS #

• 在Lambda控制台的「触发器」模块添加SQS触发器
• 选择目标队列，设置批量大小（建议10-100条，根据消息量调整）
• 配置可见性超时（需大于Lambda执行超时时间，避免重复消费）
• 开启批量触发，提升处理效率

4. 测试与监控 #

• 向SQS发送测试消息，验证Lambda是否触发、RDS是否成功写入数据
• 通过CloudWatch监控Lambda的执行日志、错误率，以及SQS的消息堆积情况
• 若出现处理失败，优先检查IAM权限、RDS网络连通性、消息格式匹配度

内容的提问来源于stack exchange，提问作者alok kumar