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

问题描述

我正在使用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","body": {
        "duration": "5510","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,player_game_status) VALUES (
            '${record.body.player_2}',${record.body.status_2}
            )`
      ),]);
  }
  );
  return{
    statusCode: 200}
  } catch (error) {
    return {
      statusCode: 400,error: error.message,};
  }
};

我已经测试了队列，并且工作正常，所以问题可能出在这里。

解决方法

问题在于，您发送给SQS的JSON被转换为字符串，因此您需要使用JSON.parse（）将其转换回Lambda中的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,player_game_status) VALUES (
            '${body.player_2}',${body.status_2}
            )`
      ),]);
    }
  );
  return{
    statusCode: 200
    
  }
  } catch (error) {
    return {
      statusCode: 400,error: error.message,};
  }
};