SQS 从队列中接收丢失的消息

Question

我设置了 AWS 基础设施，因此对 dynamo db 条目的每次更新都会在启用重复数据删除的 SQS FIFO 队列中结束。我也有一个测试覆盖这种情况，我清除队列（队列可以从套装中的其他测试获取更新。为了避免在接收正确消息之前轮询大量消息，我在运行测试之前清除队列) 并更新 Dynamo Db 并在轮询队列时检查是否收到了这些条目。这个测试很不稳定，有时它会失败，因为我发送的所有更新都没有从队列中接收到。

队列只有一个消费者，即我编写的​​测试。所以不会有另一个消费者在消费这些消息。

我通过 AWS 控制台检查了队列，在测试结束时它是空的，并且由于设置了 TIMEOUT 值而导致测试超时时不包含丢失的消息。

我在 CDK 中的队列配置

public Queue createSqsQueue() {
    return new Queue(this, "DynamoDbUpdateSqsQueue", QueueProps.builder()
            .withContentBasedDeduplication(true)
            .withFifo(true)
            .withQueueName("DynamoDbUpdateSqsQueue.fifo")
            .withReceiveMessageWaitTime(Duration.seconds(20))
            .build());
}

我的接收信息代码

private void assertExpectedDynamoDbUpdatesAreReceived() {
    List<String> expectedDynamoDbUpdates = getExpectedDynamoDbUpdates();
    List<String> actualDynamoDBUpdates = newArrayList();
    boolean allDynamoDbUpdatesReceived = false;
    stopWatch.start();
    while (!allDynamoDbUpdatesReceived && stopWatch.getTime() < TIMEOUT ) {
        List<String> receivedDynamoDbUpdates =
                AmazonSQSClientBuilder.standard().receiveMessage(queueUrl).getMessages().stream()
                        .map(this::processAndDelete)
                        .collect(Collectors.toList());
        actualDynamoDBUpdates.addAll(receivedDynamoDbUpdates);
        if(actualDynamoDBUpdates.containsAll(expectedDynamoDbUpdates)){
            allDynamoDbUpdatesReceived= true;
        }
    }
    stopWatch.stop();
    assert(allDynamoDbUpdatesReceived).isTrue();
}

Answer 1

问题不在于接收消息。它正在清理队列。根据清除队列文档 (https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-purge-queue.html)

消息删除过程最多需要 60 秒。我们推荐 无论队列大小，等待 60 秒

只需在发送更新前等待 60 秒即可解决问题。