如何阻止已删除的AWS SQS消息重复出现？（NestJS+EB+ASG场景）

我正在用Firebase Cloud Messaging（FCM）结合AWS SQS实现应用的推送通知功能，后端基于NestJS，使用@ssut/nestjs-sqs包实现SQS生产者与消费者，消息删除操作依赖@aws-sdk/client-sqs。目前遇到的问题是：消费者获取消息并发送至FCM后，过了SQS的可见性超时时间，相同的推送通知会反复发送，直到手动清空队列。怀疑是Amazon Elastic Beanstalk的自动扩展组（ASG）中多个实例重复消费同一条消息，导致消息无法正常删除。

生产者代码 #

async pushToQueue(
    devices: any[],
    title: string,
    type: string,
    additionalData: any,
    body: string
) {
    try {
        for (const device of devices) {
            await this.sqsService.send('notification', {
                id: device.id,
                body: {
                    token: device.token,
                    body,
                    title,
                    type,
                    additionalData,
                },
            });
        }
    } catch (error) {
        console.log(error);
    }
}

消费者代码 #

@SqsMessageHandler('notification', false)
public async handleNotification(message: Message): Promise<void> {
    try {
        const {
            Body,
            ReceiptHandle,
        } = message;
        const data = JSON.parse(Body);

        const {
            token,
            body,
            title,
            type,
            additionalData,
        } = data;
        await firebase
            .messaging()
            .send({
                token,
                data: {
                    type,
                    additionalData: JSON.stringify(additionalData)
                },
                notification: {
                    body,
                    title
                },
            });

        await this.sqsService.sqs.deleteMessage({
            QueueUrl: process.env.AWS_NOTIFICATION_QUEUE_URL,
            ReceiptHandle
        });
    } catch (error) {
        console.error(JSON.stringify(error, null, 2));
    }
}

排查与解决方法 #

1. 验证消息删除逻辑的有效性 #

当前代码中，删除消息的操作和FCM推送放在同一个try块里，但如果删除操作失败（比如ReceiptHandle失效、队列URL配置错误），不会有针对性的错误反馈和处理。建议给删除操作单独添加异常捕获，明确排查删除失败的原因：

// 替换原有的删除代码
try {
    await this.sqsService.sqs.deleteMessage({
        QueueUrl: process.env.AWS_NOTIFICATION_QUEUE_URL,
        ReceiptHandle
    });
    console.log(`消息已成功删除，ReceiptHandle: ${ReceiptHandle.slice(0, 20)}...`);
} catch (deleteError) {
    console.error(`删除消息失败: ${JSON.stringify(deleteError, null, 2)}`);
    // 若删除失败，抛出异常让SQS重新将消息入队（根据业务需求调整）
    throw deleteError;
}

2. 解决ASG实例重复消费问题 #

• 调整可见性超时时间：如果FCM推送加删除消息的总耗时超过了SQS的可见性超时，其他ASG实例会重新拉取这条消息。建议将可见性超时设置为单次处理耗时的2-3倍（比如处理一条消息需要5秒，就设置15秒）。
• 添加消息去重标识：在生产者发送消息时，给每条消息添加唯一ID（比如利用device.id或UUID），消费时先检查该ID是否已处理过（可以存在Redis或数据库中），如果已处理则直接删除消息，不再发送FCM。
• 检查消费者配置：确认@ssut/nestjs-sqs的消费者没有开启重复拉取机制，每个ASG实例的消费者都是独立拉取消息。

3. 切换为SQS FIFO队列（可选） #

如果业务场景允许，将标准队列替换为FIFO队列：

• 开启内容去重或手动指定MessageDeduplicationId，从根源避免重复消息。
• 通过MessageGroupId保证同组消息按顺序处理，不会被多个实例同时消费。

4. 确认环境变量一致性 #

检查所有ASG实例的AWS_NOTIFICATION_QUEUE_URL环境变量是否完全一致，避免因队列URL错误导致删除消息失败，进而让消息重新回到队列。

内容的提问来源于stack exchange，提问作者Darambazar