RabbitMQ批量消费消息解决方案

Question

我使用RabbitMQ作为不同消息的队列。当我使用来自一个队列的两个不同消费者的消息时，我会处理它们并将处理结果插入数据库：

def consumer_callback(self, channel, delivery_tag, properties, message):
    result = make_some_processing(message)
    insert_to_db(result)
    channel.basic_ack(delivery_tag)

我想大量使用队列中的消息，这将减少数据库负载。由于RabbitMQ不支持消费者批量读取消息，我将这样做smth：

some_messages_list = []
def consumer_callback(self, channel, delivery_tag, properties, message):
    some_messages_list.append({delivery_tag: message})
    if len(some_messages_list) > 1000:
        results_list = make_some_processing_bulk(some_messages_list)
        insert_to_db_bulk(results_list)
        
        for tag in some_messages_list:
            channel.basic_ack(tag)
        some_messages_list.clear()

1. 消息在全部完全处理之前处于队列中
2. 如果消费者跌倒或断开连接 - 消息保持安全

你认为这个解决方案怎么样？如果可以的话，如果消费者摔倒了，我怎样才能重新得到所有未识别的信息？

Answer 1

我已经测试这个解决方案几个月了，可以说它相当不错。直到AMPQ不提供批量消费的功能，我们不得不使用这样的一些漫游。

注意：如果您决定使用此解决方案，请注意使用多个消费者（线程）进行并发消费，或者使用一些锁（我使用过pythonthreading.Lock模块）来保证没有竞争条件发生。