SQS 消息在队列中空闲长达 99 秒，直到 readMessage 将其拾取

Question

我有一个从 s3 转换文件的 ETL 进程。 S3 事件通知向 SQS 发送消息，并且 Fargate 任务轮询队列以获取消息。我的问题是消息在不定的时间内没有被接收到，最多 99 秒。

我*每 2 秒查找一次消息。我可以通过控制台确认消息在队列中，但很长一段时间都没有被读取。此外，我正在打印消息在日志队列中的延迟，基于 MessageSystemAttributeNameApproximateReceiveCount 和 MessageSystemAttributeNameSentTimestamp 之间的差异

    queueURL := os.Getenv("QUEUE_URL")
    csvBucket := os.Getenv("CSV_BUCKET_NAME")

    sess, _ := session.NewSession()
    //TODO: check and handle error
    svc := sqs.New(sess)

    for {
        time.Sleep(time.Duration(2) * time.Second)

        msgResult, err := svc.ReceiveMessage(&sqs.ReceiveMessageInput{
            AttributeNames: []*string{
                aws.String(sqs.MessageSystemAttributeNameSenderId),
                aws.String(sqs.MessageSystemAttributeNameSentTimestamp),
                aws.String(sqs.MessageSystemAttributeNameApproximateReceiveCount),
                aws.String(sqs.MessageSystemAttributeNameApproximateFirstReceiveTimestamp),
                aws.String(sqs.MessageSystemAttributeNameSequenceNumber),
                aws.String(sqs.MessageSystemAttributeNameMessageDeduplicationId),
                aws.String(sqs.MessageSystemAttributeNameMessageGroupId),
                aws.String(sqs.MessageSystemAttributeNameAwstraceHeader),
            },
            MessageAttributeNames: []*string{
                aws.String(sqs.QueueAttributeNameAll),
            },
            QueueUrl:            aws.String(queueURL),
            MaxNumberOfMessages: aws.Int64(10),
            VisibilityTimeout:   aws.Int64(30), // 60 seconds
            WaitTimeSeconds:     aws.Int64(0),
        })

        if err != nil {
            log.Printf("recieve message failed: %v", err)
            return
        }

        if len(msgResult.Messages) == 0 {
            continue
        }
        //TODO: process all results
        // fmt.Printf("Success: %+v\n", msgResult.Messages)
        log.Printf("amount of messages %v", len(msgResult.Messages))

        i, err := strconv.ParseInt(*msgResult.Messages[0].Attributes["ApproximateFirstReceiveTimestamp"], 10, 64)
        if err != nil {
            log.Printf("failed to parse int: %v", err)
            return
        }
        tm1 := time.Unix(0, i*int64(1000000))

        ii, err := strconv.ParseInt(*msgResult.Messages[0].Attributes["SentTimestamp"], 10, 64)
        if err != nil {
            panic(err)
        }
        tm2 := time.Unix(0, ii*int64(1000000))

/* some code to download from s3 and upload to another bucket */

        _, err = svc.DeleteMessage(&sqs.DeleteMessageInput{
            QueueUrl:      aws.String(queueURL),
            ReceiptHandle: msgResult.Messages[0].ReceiptHandle,
        })
        if err != nil {
            log.Println("delete failed")
            continue
        }
        log.Println("message delete succeeded")
    }
        

timestamp,message
1599145270790,2020/09/03 15:01:10 seconds the message was unprocessed in the queue: 0.909
1599145426542,2020/09/03 15:03:46 seconds the message was unprocessed in the queue: 19.835
1599145472884,2020/09/03 15:04:32 seconds the message was unprocessed in the queue: 4.721
1599145611897,2020/09/03 15:06:51 seconds the message was unprocessed in the queue: 24.793
1599145720293,2020/09/03 15:08:40 seconds the message was unprocessed in the queue: 40.296
1599145930662,2020/09/03 15:12:10 seconds the message was unprocessed in the queue: 66.736
1599145997155,2020/09/03 15:13:17 seconds the message was unprocessed in the queue: 10.961
1599146249316,2020/09/03 15:17:29 seconds the message was unprocessed in the queue: 99.084
1599146319998,2020/09/03 15:18:39 seconds the message was unprocessed in the queue: 35.015
1599146361620,2020/09/03 15:19:21 seconds the message was unprocessed in the queue: 17.353
1599146438448,2020/09/03 15:20:38 seconds the message was unprocessed in the queue: 45.878

我很迷失自己做错了什么，然后我将其上传到另一个存储桶到队列中，它被 lambda 拾取，始终在大约一秒钟内完成

定义管道的cdk代码

        super(scope, id)

        const dlq = new sqs.Queue(this, 'DeadLetterQueue', {
            retentionPeriod: cdk.Duration.days(14),
        });

        const q = new sqs.Queue(this, 'ProcessingQueue', {
            deadLetterQueue: {
                queue: dlq,
                maxReceiveCount: 3,
            },
        });

        const queueFargate = new ecsPatterns.QueueProcessingFargateService(this, 'Service', {
            queue: q,
            vpc: props.vpc,
            memoryLimitMiB: 512,
            cpu: 256,
            image: ecs.ContainerImage.fromAsset('services/convert'),
            platformVersion: ecs.FargatePlatformVersion.VERSION1_4,
            desiredTaskCount: 1,
            environment: {
                QUEUE_URL: q.queueUrl,
                CSV_BUCKET_NAME: props.convertBucket.bucketName,
            },
        });

        props.uploadBucket.addEventNotification(s3.EventType.OBJECT_CREATED, new s3n.SqsDestination(queueFargate.sqsQueue))
        props.uploadBucket.grantRead(queueFargate.service.taskDefinition.taskRole)
        props.convertBucket.grantWrite(queueFargate.service.taskDefinition.taskRole)

适用于 Go 的 AWS 开发工具包的版本？

github.com/aws/aws-sdk-go v1.34.6

去 1.14

预期行为

消息一进入队列就应该被读取。

Answer 1

这可能是使用短轮询而不是长轮询的结果。根据Amazon SQS short and long polling 上的 AWS 文档：

通过短轮询，ReceiveMessage 请求仅查询服务器的子集（基于加权随机分布）以查找可包含在响应中的消息。即使查询未找到任何消息，Amazon SQS 也会立即发送响应。

另一方面，长轮询将：

通过查询所有（而不是部分）Amazon SQS 服务器来减少错误的空响应。

您可以通过为WaitTimeSeconds 传递一个非零值并在每次迭代中移除两秒休眠来在代码中实现长轮询。