如何制作 x 小时的 SQS 延迟队列

Question

我需要一个队列来处理延迟 x 小时后的消息。而且我需要一种数据驱动的所有基于事件的方法，不使用任何调度程序等。

场景是我将一些实时数据发送到 SNS 主题，然后从那里发送到不同的 SQS 队列，以供不同的 AWS Lambda 函数使用。

其中一个 Lambda 函数需要在延迟 3 小时后处理消息。但是，最长交货延迟为 15 分钟。如果我第一次阅读该消息，它将自动从 SQS 中删除，因为我正在使用事件源映射触发器来调用 lambda 函数。

所以，我想知道如何避免删除消息并使其在第一次处理时不可见？

任何想法/帮助将不胜感激。

Answer 1

Amazon SQS 不会执行您的请求。另外，我不建议使用任何“技巧”来强制延迟。

我建议您查看 AWS Step Functions。它可以协调 AWS Lambda 函数之间的交互，并且可以配置为在调用 AWS Lambda 函数之前等待（睡眠）一段时间。

Answer 2

我测试过了。这似乎是可行的。我使用下面的代码进行测试。但是，这似乎不是做您想要实现的目标的“良好实践”方式。我看到两个主要问题：

1. 机上按摩次数上限为 120,000。因此，您几乎可以无限扩展您的 SQS 队列。

2. 您的指标将充满错误调用，并且很难区分真正失败的调用和故意失败的调用。

因此，我会研究其他解决方案。

import json
import os
import time

import boto3

sqs = boto3.client('sqs')

queue_url = os.environ['QUEUE_URL']

new_visibility_timeout = 120 # seconds

def lambda_handler(event, context):

    print(json.dumps(event))

    current_time = time.time()

    no_of_new_records = 0

    for record in event['Records']:

        msg_timestamp = float(record['attributes']['SentTimestamp'])/1000

        msg_age = current_time - msg_timestamp

        print(f"Message age: {msg_age} seconds")

        if msg_age > new_visibility_timeout:

            print("Message to be successfully processed and deleted from queue")

            response = sqs.delete_message(
                QueueUrl=queue_url,
                ReceiptHandle=record['receiptHandle']
            )

            print(response)

        else:

            print("Set long visibility timeout")

            response = sqs.change_message_visibility(
                QueueUrl=queue_url,
                ReceiptHandle=record['receiptHandle'],
                VisibilityTimeout=new_visibility_timeout
            )

            print(response)

            no_of_new_records += 1

    if no_of_new_records > 0:
        raise Exception("Fail the lambda")

    return {'statusCode': 200}