使用 Lambda 将消息从 SQS 写入数据库

Question

我正在使用 AWS 开展一个项目，并创建了一个我已连接到的端点和 SQS 队列，该队列随后会触发一个 Lambda 函数。 Lambda 应该将数据写入数据库，但由于某种原因它没有。当我尝试使用这样的请求手动触发它时

{
  "Records": [
    {
      "messageId": "11d6ee51-4cc7-4302-9e22-7cd8afdaadf5",
      "receiptHandle": "AQEBBX8nesZEXmkhsmZeyIE8iQAMig7qw...",
      "body": {
        "duration": "1230",
        "player_1": "UR-da336be50ba9b8e53b8",
        "player_2": "UR-a67322a021284404128",
        "status_1": 1,
        "status_2": 0
      },
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1573251510774",
        "SequenceNumber": "18849496460467696128",
        "MessageGroupId": "1",
        "SenderId": "AIDAIO23YVJENQZJOL4VO",
        "MessageDeduplicationId": "1",
        "ApproximateFirstReceiveTimestamp": "1573251510774"
      },
      "messageAttributes": {},
      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:fifo.fifo",
      "awsRegion": "us-east-2"
    },
    {
      "messageId": "11d6ee51-4cc7-4302-9e22-7cd8afdaadf5",
      "receiptHandle": "AQEBBX8nesZEXmkhsmZeyIE8iQAMig7qw...",
      "body": {
        "duration": "5510",
        "player_1": "UR-da336be50ba9b8e53b8",
        "player_2": "UR-a67322a021284404128",
        "status_1": 1,
        "status_2": 0
      },
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1573251510774",
        "SequenceNumber": "18849496460467696128",
        "MessageGroupId": "1",
        "SenderId": "AIDAIO23YVJENQZJOL4VO",
        "MessageDeduplicationId": "1",
        "ApproximateFirstReceiveTimestamp": "1573251510774"
      },
      "messageAttributes": {},
      "md5OfBody": "e4e68fb7bd0e697a0ae8f1bb342846b3",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-2:123456789012:fifo.fifo",
      "awsRegion": "us-east-2"
    }
  ]
}

它工作正常，但是当它从 SQS 调用时没有任何反应，它会从 SQS 中删除消息。

我的 Lambda 代码：

const { Client } = require("pg");

const client = new Client({
  user: process.env.POSTGRES_USER,
  host: process.env.POSTGRES_HOST,
  database: process.env.POSTGRES_DATABASE,
  password: process.env.POSTGRES_PASSWORD,
  port: parseInt(process.env.POSTGRES_PORT),
});

client.connect();

async function asyncForEach(array, callback) {
  for (let index = 0; index < array.length; index++) {
    await callback(array[index], index, array);
  }
}

exports.handler = async (event) => {
  try {
    await asyncForEach(event.Records, async (record) => {
      //Writes the game info to the DB
    const result = await client.query(
      `INSERT INTO game_data (duration) VALUES (
        ${record.body.duration}
        ) RETURNING game_id`
    );

    const res = await Promise.all([
      client.query(
        `INSERT INTO player_game_data (user_id,game_id,player_game_status) VALUES (
            '${record.body.player_1}',
            '${result.rows[0].game_id}',
             ${record.body.status_1}
            )`
      ),
      client.query(
        `INSERT INTO player_game_data (user_id,game_id,player_game_status) VALUES (
            '${record.body.player_2}',
            '${result.rows[0].game_id}',
             ${record.body.status_2}
            )`
      ),
    ]);
  }
  );
  return{
    statusCode: 200}
  } catch (error) {
    return {
      statusCode: 400,
      error: error.message,
    };
  }
};

我已经测试了队列，它工作正常，所以问题可能在这里..

Answer 1

问题是您发送到 SQS 的 JSON 被转换为字符串，因此您需要在 Lambda 中使用 JSON.parse() 将其转换回 JSON。

通过 SQS 发出的请求如下所示：

{
  "Records": [
    {
      "messageId": "bc62a976-06ef-4727-8bb5-8d7b0a474f7d",
      "receiptHandle": "AQEBd4oxuMTytPn8AWORy992aGqGO5By+pM1x2dtZpyn0n8cxTJEd9/BXemUnAAbU+tx1jRlsRWCYhPnrrBvCj91nUpw5gT10WGkuQcv6fCH+ePqqON6sIHy9+8csqhzCwphDqdA23SLfidEGMwuW8mvNN+Lh541vfgHoYSQhMv51qLjHADbiSUzfsIYVnvmqU+C3D55OX/OhDOJoWY87XIEjpSEqRKx4s8wTF6edpYyun0IBYUA68W5CFkg+RBuWPeKsGLNENCvCpawcknYOCKrxeMrWRTh73qHZzH6QnNTO5S4fzQONKH2MWjFsIy7T01w1feNSD3qt/m3vakWhQnhi8VDn9KUJCIdKbhxpdqZB3QSPAKvfjRtEkwXQu2pGUpezMtWbNmsQfaEw84+7BV/CQ==",
      "body": "{\r\n    \"duration\": \"69\",\r\n    \"player_1\": \"UR-da336be50ba9b8e53b8\",\r\n    \"player_2\": \"UR-a67322a021284404128\",\r\n    \"status_1\": 0,\r\n    \"status_2\": 1\r\n}",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "AWSTraceHeader": "Root=1-5f454d30-8772a1ac004584ac5e0cbf48",
        "SentTimestamp": "1598377264745",
        "SenderId": "AROAYLJPJR5FJH6OQCRDF:BackplaneAssumeRoleSession",
        "ApproximateFirstReceiveTimestamp": "1598377264750"
      },
      "messageAttributes": {},
      "md5OfBody": "0be85f29f6c6dd29e328b58a01e3db2a",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:aws:sqs:us-east-1:574014197578:dev-Post-GameResults",
      "awsRegion": "us-east-1"
    }
  ]
}

在我解析完正文后，一切都按预期工作..

编辑后的代码：

const { Client } = require("pg");

const client = new Client({
  user: process.env.POSTGRES_USER,
  host: process.env.POSTGRES_HOST,
  database: process.env.POSTGRES_DATABASE,
  password: process.env.POSTGRES_PASSWORD,
  port: parseInt(process.env.POSTGRES_PORT),
});

client.connect();

async function asyncForEach(array, callback) {
  for (let index = 0; index < array.length; index++) {
    await callback(array[index], index, array);
  }
}

exports.handler = async (event) => {
  try {
      await asyncForEach(event.Records, async (record) => {
        var body = JSON.parse(record.body);
      //Writes the game info to the DB
    const result = await client.query(
      `INSERT INTO game_data (duration) VALUES (${body.duration}) RETURNING game_id`
    );

    const res = await Promise.all([
      client.query(
        `INSERT INTO player_game_data (user_id,game_id,player_game_status) VALUES (
            '${body.player_1}',
            '${result.rows[0].game_id}',
             ${body.status_1}
            )`
      ),
      client.query(
        `INSERT INTO player_game_data (user_id,game_id,player_game_status) VALUES (
            '${body.player_2}',
            '${result.rows[0].game_id}',
             ${body.status_2}
            )`
      ),
    ]);
    }
  );
  return{
    statusCode: 200
    
  }
  } catch (error) {
    return {
      statusCode: 400,
      error: error.message,
    };
  }
};