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

Question

我已经建立了AWS架构体系，以便每次对发电机数据库条目的更新都以启用重复数据删除的SQS先进先出队列结束。我还有一个测试来覆盖这个场景，在那里我清除了队列（队列可以从套装中的其他测试中获得更新。为了避免在收到正确的消息之前必须轮询大量的消息，我在运行测试之前清除队列），更新Dynamo Db，并在轮询队列时检查这些条目是否收到。这个测试是不稳定的，有时会失败，因为我发送的所有更新都没有从队列中收到。

队列只有一个消费者，这就是我编写的测试。所以，这并不是说有另一个消费者在消费这些信息。

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

我在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秒，即可解决此问题。