RabbitMQ批量消费方案可行性评估及故障后未确认消息重获取咨询

问题1：上述批量消费实现方案是否合理可行？ #

该方案的核心思路是可行的，但存在多处需要修正的设计缺陷和遗漏配置，无法直接满足你提出的两个约束：

• 代码逻辑bug：你当前将每个消息封装为单key字典存入some_messages_list，后续遍历列表确认消息时，直接for tag in some_messages_list取到的是字典对象而非delivery_tag，会导致ack调用报错，消息无法被正常确认。你可以改为把消息和tag存为元组，比如some_messages_list.append( (delivery_tag, message) )，后续批量处理和ack的时候遍历取第一个元素作为tag即可。
• 缺少prefetch_count配置：如果没有给消费者通道设置channel.basic_qos(prefetch_count=xxx)，RabbitMQ会尽可能将队列中的所有消息推送到消费者本地缓存，不仅会导致你内存中攒的消息远超过1000条，还可能把消费者进程撑爆。建议将prefetch_count设置为略大于你的批量阈值1000，比如1200，既可以保证你能攒够批量处理的消息量，也不会让消费者缓存过多消息。
• 缺少超时兜底逻辑：如果消息流量低，长时间凑不够1000条，本地缓存的消息会一直得不到处理，超过RabbitMQ的未确认消息超时时间后，消息会被自动重新入队，导致重复消费。你需要额外加一个定时任务，比如每隔30秒不管攒了多少条都强制处理一次当前缓存的消息。
• 并发安全问题：如果你的消费者是多线程/多协程运行，全局的some_messages_list没有加锁，会出现并发写入异常、丢消息的问题。
• 基础配置依赖：需要确保你的队列开启了持久化、消息发送时设置了持久化属性、消费者消费时auto_ack参数设置为False，才能保证消息不丢。

问题2：若方案可行，消费者崩溃后如何重新获取所有未确认的消息？ #

RabbitMQ本身自带未确认消息的重入队机制，不需要你额外做特殊开发：

• 只要消费者和RabbitMQ的TCP连接断开（无论是进程崩溃、网络中断还是主动关闭），该消费者所有已经下发但未调用basic_ack确认的消息，都会被RabbitMQ自动重新入队，分配给同队列的其他在线消费者。
• 如果该消费者重启后重新连接到队列，这些重新入队的消息也可以被该消费者重新消费到。
• 注意事项：该机制会导致消息可能被重复投递，所以你的业务处理逻辑和数据库写入逻辑必须实现幂等，避免重复插入数据。

内容的提问来源于stack exchange，提问作者Ярига Олег