您不能在使用者回调中对消息进行批处理，但可以使用线程安全库并使用多个线程来使用数据。这样做的好处是，您可以在五个不同的线程上获取五条消息，并在需要时组合数据。
作为一个例子，你可以看看我将如何使用我的AMQP库实现这一点。https://github.com/eandersson/amqpstorm/blob/master/examples/scalable_consumer.py

下面的代码将使用channel.consume来开始使用消息。当达到所需的消息数量时，我们将退出/停止。
我已经设置了一个batch_size来防止一次拉取大量的消息。你可以随时改变batch_size来满足你的需要。

def consume_messages(queue_name: str):
    msgs = list([])
    batch_size = 500

    q = channel.queue_declare(queue_name, durable=True, exclusive=False, auto_delete=False)
    q_length = q.method.message_count
    
    if not q_length:
        return msgs

    msgs_limit = batch_size if q_length > batch_size else q_length

    try:
        # Get messages and break out
        for method_frame, properties, body in channel.consume(queue_name):

            # Append the message
            try:
                msgs.append(json.loads(bytes.decode(body)))
            except:
                logger.info(f"Rabbit Consumer : Received message in wrong format {str(body)}")

            # Acknowledge the message
            channel.basic_ack(method_frame.delivery_tag)

            # Escape out of the loop when desired msgs are fetched
            if method_frame.delivery_tag == msgs_limit:

                # Cancel the consumer and return any pending messages
                requeued_messages = channel.cancel()
                print('Requeued %i messages' % requeued_messages)
                break

    except (ChannelWrongStateError, StreamLostError, AMQPConnectionError) as e:
        logger.info(f'Connection Interrupted: {str(e)}')

    finally:
        # Close the channel and the connection
        channel.stop_consuming()
        channel.close()

    return msgs