我从 SNS 主题订阅时出现错误的 SQS AWS 消息

Question

我对下一个设计有疑问：

当我在我的 SQS 订阅者中接收到消息时，消息的模型是错误的，例如：

{
  "Type" : "Notification",
  "MessageId" : "7a6789f0-02f0-5ed3-8a11-deebcd08f145",
  "TopicArn" : "arn:aws:sns:us-east-2:167186109795:name_sns_topic",
  "Message" : "My JSON message",
  "Timestamp" : "1987-04-23T17:17:44.897Z",
  "SignatureVersion" : "1",
  "Signature" : "string",
  "SigningCertURL" : "url",
  "UnsubscribeURL" : "url",
  "MessageAttributes" : {
    "X-Header1" : {"Type":"String","Value":"value1"},
    "X-Header2" : {"Type":"String","Value":"value2"},
    "X-Header3" : {"Type":"String","Value":"value3"},
    "X-HeaderN" : {"Type":"String","Value":"value4"}
  }
}

从SQS接收消息的通用模型应该是：

{
  "Records": [
    {
      "messageId": "19dd0b57-b21e-4ac1-bd88-01bbb068cb78",
      "receiptHandle": "MessageReceiptHandle",
      "body": "Hello from SQS!",
      "attributes": {
        "ApproximateReceiveCount": "1",
        "SentTimestamp": "1523232000000",
        "SenderId": "123456789012",
        "ApproximateFirstReceiveTimestamp": "1523232000001"
      },
      "messageAttributes": {},
      "md5OfBody": "7b270e59b47ff90a553787216d55d91d",
      "eventSource": "aws:sqs",
      "eventSourceARN": "arn:{partition}:sqs:{region}:123456789012:MyQueue",
      "awsRegion": "{region}"
    }
  ]
}

在我的处理程序中，Java Lambda（示例代码）抛出异常，因为收到的 de 消息的结构不是 SQS 事件：

public class MyHandler implements RequestHandler<SQSEvent, String> {

  @Override
  public String handleRequest(SQSEvent event, Context context) {
    LambdaLogger logger = context.getLogger();

    for (SQSEvent.SQSMessage msg : event.getRecords()) {
      logger.log("SQS message body: " + msg.getBody());
      logger.log("Get attributes: " + msg.getMessageAttributes().toString());

      msg.getMessageAttributes()
              .forEach(
                      (k, v) -> {
                        logger.log("key: " + k + "value: " + v.getStringValue());
                      });
    }
    return "Successful";
  }
}

How can I do for handle the message thats its receiving ?

Answer 1

在我看来，这并没有很好地记录下来，但是一旦你弄清楚它就不错了。

第一件事是我不使用预定义的 Lambda 对象。我将所有内容读入字符串并从那里获取。所以我的Lamda函数的基础是：

public void handleRequest(InputStream inputStream, OutputStream outputStream, Context context) throws IOException {
    // copy InputStream to String, avoiding 3rd party libraries
    ByteArrayOutputStream result = new ByteArrayOutputStream();
    byte[] buffer = new byte[1024];
    int length;
    while ((length = inputStream.read(buffer)) != -1) {
        result.write(buffer, 0, length);
    }

    String jsonString = result.toString();
}

当您“直接”从 SNS 到 Lambda 时，消息看起来像（为了长度而删除了一些字段）：

{
    "Records": [
        {
            "EventSource": "aws:sns",
            "EventVersion": "1.0",
            "Sns": {
                "Type": "Notification",
                "Subject": "the message subject",
                "Message": "{\"message\": \"this is the message\", \"value\": 100}",
                "Timestamp": "2020-04-24T21:44:28.220Z",
                "SignatureVersion": "1"
            }
        }
    ]
}

我发送了一条带有两个简单字段的 JSON 测试消息。使用JsonPath 读取所有内容中的“消息”字段：

String snsMessage = JsonPath.read(jsonString, "$.Records[0].Sns.Message");
String realMessage = JsonPath.read(snsMessage, "$.message");

但是当它转到 SNS -> SQS -> Lambda（或者，实际上是任何 SNS -> SQS 路径）时，SNS 消息现在大部分都被包装和转义在 SQS 消息中：

{
    "Records": [
        {
            "messageId": "ca8c53e5-8417-4479-a720-d4ecf970ca68",
            "body": "{\n  \"Type\" : \"Notification\",\n  \"Subject\" : \"the message subject\",\n  \"Message\" : \"{\\\"message\\\": \\\"this is the message\\\", \\\"value\\\": 100}\"\n}",
            "attributes": {
                "ApproximateReceiveCount": "1"
            },
            "md5OfBody": "6a4840230aca6a7bf7934bf191a529b8",
            "eventSource": "aws:sqs"
        }
    ]
}

所以在这种情况下，该值在 Records[0].body 中，但其中包含另一个 JSON 对象。我承认可能有一种更简单的方法，但据我发现我必须解析 3 次：

String sqsBody = <as read in lambda>;
String recordBody = JsonPath.read(sqsBody, "$.Records[0].body");
String internalMessage = JsonPath.read(recordBody, "$.Message");
// now read out of the sns message
String theSnsMessage = JsonPath.read(message, "$.message");