如何解决使用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,};
}
};
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。