MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

消息队列的消息大小限制与拆分策略

2024-03-124.1k 阅读

消息队列的消息大小限制

常见消息队列的消息大小限制

不同的消息队列产品对消息大小有着各自不同的限制，这是由其设计理念、底层存储机制以及网络传输等多方面因素决定的。

  1. RabbitMQ：RabbitMQ 的默认消息大小限制通常在 100KB 到 2MB 之间。这一限制主要是基于其 AMQP 协议以及内存使用等方面的考虑。在 RabbitMQ 中，消息会先存储在内存中，当内存达到一定阈值后才会持久化到磁盘。如果消息过大，会占用大量内存，可能导致内存不足，进而影响整个 RabbitMQ 服务器的性能。
  2. Kafka：Kafka 的消息大小限制相对较为灵活。默认情况下，Kafka 可以支持单个消息最大为 1MB。不过，通过配置参数 message.max.bytes（broker 端）和 max.request.size（producer 端），可以将这个限制提高到数 MB 甚至更大。Kafka 的设计理念侧重于高吞吐量和低延迟，它采用了基于磁盘的存储机制，消息会顺序写入磁盘，因此对消息大小的限制不像 RabbitMQ 那样严格受限于内存。
  3. RocketMQ：RocketMQ 支持的消息大小默认上限为 4MB。RocketMQ 在设计上充分考虑了分布式场景下的消息处理，它的存储机制和通信协议决定了这个消息大小限制。在实际应用中，如果消息大小超过这个限制，就需要采取相应的处理策略。

消息大小限制的影响因素

  1. 内存与存储：对于像 RabbitMQ 这类先将消息存储在内存中的消息队列，消息大小直接关系到内存的占用。过大的消息会迅速耗尽内存，导致服务器性能下降甚至崩溃。而对于基于磁盘存储的 Kafka 和 RocketMQ，虽然可以存储较大消息，但过大的消息也会影响磁盘 I/O 性能，因为磁盘读写存在一定的块大小和传输效率限制。
  2. 网络传输：消息队列通常会在分布式环境中使用，消息需要通过网络进行传输。网络带宽和 MTU（最大传输单元）限制了一次能够传输的数据量。如果消息大小超过了网络所能承载的范围，就会导致网络传输错误或分片传输，增加传输延迟和复杂性。例如，在常见的以太网环境中，MTU 一般为 1500 字节，如果消息过大，就需要进行分片处理，这会增加网络层的开销。
  3. 处理性能：消息队列中的消息最终需要被消费者处理。如果消息过大，消费者在处理时可能需要耗费更多的资源，如 CPU 用于解析和处理数据，内存用于存储消息内容。这可能导致消费者处理速度变慢，甚至出现处理超时等问题，影响整个系统的性能和稳定性。

消息拆分策略

基于固定大小拆分

  1. 原理：基于固定大小拆分策略是将大消息按照固定的字节数进行拆分。例如，设定一个固定的拆分大小为 100KB，当消息大小超过这个值时，就将其拆分成多个 100KB 的小消息。这种策略简单直观，易于实现和理解。
  2. 优点：实现简单，不需要对消息内容进行复杂的解析。拆分后的小消息大小一致，便于在消息队列中进行管理和传输，也有利于消费者进行统一的处理。
  3. 缺点：可能会造成一定的空间浪费。如果消息大小不是固定拆分大小的整数倍，最后一个小消息可能会存在大量的空余空间。例如，消息大小为 250KB，按照 100KB 拆分，会得到 3 个小消息，其中最后一个小消息只有 50KB，浪费了 50KB 的空间。
  4. 代码示例（以 Python 和 RabbitMQ 为例）：
import pika
import math


# 拆分消息函数
def split_message(message, split_size):
    message_parts = []
    total_size = len(message)
    num_parts = math.ceil(total_size / split_size)
    for i in range(num_parts):
        start = i * split_size
        end = min((i + 1) * split_size, total_size)
        message_parts.append(message[start:end])
    return message_parts


# 连接 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='large_message_queue')

# 假设这是一个大消息
large_message = b'x' * 300000

# 拆分消息
split_messages = split_message(large_message, 100000)

# 发送拆分后的消息
for part in split_messages:
    channel.basic_publish(exchange='', routing_key='large_message_queue', body=part)

print("消息已拆分并发送")

connection.close()

基于数据结构拆分

  1. 原理：当消息是一种结构化的数据，如 JSON 数组或 XML 文档时，可以根据数据结构进行拆分。例如，对于一个包含大量元素的 JSON 数组，可以按照一定数量的元素进行拆分。假设 JSON 数组中有 1000 个元素，设定每 100 个元素拆分成一个小消息。
  2. 优点：能够充分利用消息的内部结构，避免固定大小拆分可能产生的空间浪费问题。同时，由于拆分后的小消息仍然保持一定的逻辑完整性，消费者在处理时可以更容易地恢复原始数据结构。
  3. 缺点：实现相对复杂，需要对消息的数据结构有深入的了解。并且，不同的数据结构可能需要不同的拆分逻辑，通用性较差。
  4. 代码示例（以 Python 和 Kafka 为例，假设消息是 JSON 数组）：
from kafka import KafkaProducer
import json


# 拆分 JSON 数组消息函数
def split_json_array_message(message, elements_per_part):
    data = json.loads(message)
    message_parts = []
    num_parts = math.ceil(len(data) / elements_per_part)
    for i in range(num_parts):
        start = i * elements_per_part
        end = min((i + 1) * elements_per_part, len(data))
        part = data[start:end]
        message_parts.append(json.dumps(part).encode('utf-8'))
    return message_parts


# 创建 Kafka producer
producer = KafkaProducer(bootstrap_servers=['localhost:9092'])

# 假设这是一个大的 JSON 数组消息
large_json_message = '[{"id": 1, "name": "item1"}, {"id": 2, "name": "item2"}]' * 1000

# 拆分消息
split_messages = split_json_array_message(large_json_message, 100)

# 发送拆分后的消息
for part in split_messages:
    producer.send('large_message_topic', value=part)

producer.flush()
print("消息已拆分并发送")

基于业务逻辑拆分

  1. 原理：根据消息所承载的业务逻辑进行拆分。例如，在一个电商订单处理系统中，如果一个订单消息包含多个商品信息以及配送、支付等信息，并且不同的业务模块对这些信息的处理方式不同，可以按照业务模块进行拆分。将商品信息部分拆分成一个消息发送给商品库存管理模块，将支付信息部分拆分成另一个消息发送给支付处理模块。
  2. 优点：能够更好地满足不同业务模块的需求，提高系统的可扩展性和灵活性。同时，由于拆分后的消息针对性更强，各个业务模块的处理效率可能会得到提升。
  3. 缺点：对业务逻辑的依赖性强，需要深入了解业务需求和流程。如果业务逻辑发生变化，拆分策略可能需要进行相应的调整，维护成本较高。
  4. 代码示例（以 Java 和 RocketMQ 为例）：
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;


public class BusinessLogicSplitExample {
    public static void main(String[] args) throws Exception {
        // 创建 RocketMQ producer
        DefaultMQProducer producer = new DefaultMQProducer("business_logic_split_group");
        producer.setNamesrvAddr("localhost:9876");
        producer.start();

        // 假设这是一个包含业务信息的大消息
        String largeBusinessMessage = "商品信息: 手机, 数量 1; 支付信息: 微信支付, 金额 5000; 配送信息: 北京";

        // 按照业务逻辑拆分消息
        String[] parts = largeBusinessMessage.split(";");
        for (String part : parts) {
            Message message = new Message("business_message_topic", part.getBytes());
            SendResult sendResult = producer.send(message);
            System.out.println("发送结果: " + sendResult);
        }

        producer.shutdown();
    }
}

消息拆分后的重组

基于消息编号重组

  1. 原理：在拆分消息时，为每个小消息分配一个唯一的编号，并记录消息的总数。消费者在接收到小消息后，根据消息编号对其进行排序，然后按照编号顺序将小消息组合成原始的大消息。例如，将一个大消息拆分成 5 个小消息，分别编号为 1 - 5，消费者收到这些小消息后，先按照编号排序，再将它们拼接起来。
  2. 优点：实现相对简单，只需要在拆分和重组时维护好消息编号即可。适用于各种类型的拆分策略，无论是固定大小拆分、基于数据结构拆分还是基于业务逻辑拆分。
  3. 缺点：如果消息在传输过程中丢失或编号出现错误，可能导致重组失败。因此，需要在消息传输过程中保证消息编号的准确性和完整性，通常可以通过增加校验和等方式来解决。
  4. 代码示例（以 Python 和 RabbitMQ 为例，基于固定大小拆分后的重组）：
import pika


# 连接 RabbitMQ
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 声明队列
channel.queue_declare(queue='large_message_queue')

received_messages = {}


def callback(ch, method, properties, body):
    message_id = properties.headers.get('message_id')
    total_parts = properties.headers.get('total_parts')
    received_messages[message_id] = body
    if len(received_messages) == total_parts:
        sorted_messages = [received_messages[i] for i in range(total_parts)]
        original_message = b''.join(sorted_messages)
        print("重组后的消息: ", original_message)


# 消费消息
channel.basic_consume(queue='large_message_queue', on_message_callback=callback, auto_ack=True)

print('等待接收消息...')
channel.start_consuming()

基于数据结构重组

  1. 原理：当消息是按照数据结构进行拆分时，消费者可以根据数据结构的特点进行重组。例如，对于按照 JSON 数组元素拆分的消息，消费者在接收到所有拆分后的小消息后，将其中的 JSON 数组元素合并成一个完整的 JSON 数组。
  2. 优点：由于是基于数据结构本身进行重组，不需要额外维护复杂的编号等信息。重组过程相对直观，只要数据结构在拆分和重组过程中保持一致，就能够准确地恢复原始消息。
  3. 缺点：对数据结构的依赖性强，如果数据结构发生变化，重组逻辑也需要相应调整。同时，如果拆分后的小消息丢失或数据结构不完整，可能导致重组失败。
  4. 代码示例（以 Python 和 Kafka 为例，基于 JSON 数组拆分后的重组）：
from kafka import KafkaConsumer
import json


# 创建 Kafka consumer
consumer = KafkaConsumer('large_message_topic', bootstrap_servers=['localhost:9092'])

received_parts = []

for message in consumer:
    part = json.loads(message.value.decode('utf-8'))
    received_parts.extend(part)

original_json_message = json.dumps(received_parts)
print("重组后的 JSON 消息: ", original_json_message)

基于业务逻辑重组

  1. 原理：在基于业务逻辑拆分的情况下，消费者根据业务逻辑将拆分后的消息进行重组。例如，在电商订单处理系统中，商品库存管理模块收到商品信息消息，支付处理模块收到支付信息消息，这些模块在各自处理完自身业务后，可能需要将处理结果发送给订单汇总模块，订单汇总模块根据业务规则将这些结果重组为完整的订单处理结果。
  2. 优点：紧密结合业务逻辑，能够更好地满足业务需求。通过在不同业务模块之间传递和处理拆分后的消息，可以实现复杂的业务流程。
  3. 缺点：业务逻辑复杂，重组过程可能涉及多个模块之间的交互和协调。如果业务流程发生变化，重组逻辑的调整难度较大，维护成本高。
  4. 代码示例（以 Java 和 RocketMQ 为例，基于业务逻辑拆分后的重组）：
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;


public class BusinessLogicRecombineExample {
    public static void main(String[] args) throws Exception {
        // 创建 RocketMQ consumer
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("business_logic_recombine_group");
        consumer.setNamesrvAddr("localhost:9876");
        consumer.subscribe("business_message_topic", "*");

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                // 这里假设已经接收到所有基于业务逻辑拆分的消息
                // 按照业务逻辑进行重组
                StringBuilder originalMessage = new StringBuilder();
                for (MessageExt msg : msgs) {
                    originalMessage.append(new String(msg.getBody())).append(";");
                }
                System.out.println("重组后的业务消息: " + originalMessage.toString());
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        consumer.start();
        System.out.println("Consumer 已启动");
    }
}

消息大小限制与拆分策略的优化

调整消息队列配置

  1. 增加消息大小限制：对于一些消息队列产品，如 Kafka 和 RocketMQ，可以通过调整配置参数来增加消息大小限制。在 Kafka 中，可以通过修改 message.max.bytes 和 max.request.size 参数来提高允许的消息大小。在 RocketMQ 中，也可以通过相应的配置文件修改来调整消息大小上限。不过，在增加消息大小限制时，需要充分考虑服务器的内存、磁盘和网络等资源情况，避免因消息过大导致系统性能问题。
  2. 优化存储与传输配置：例如，在 RabbitMQ 中，可以调整内存阈值和磁盘存储策略，使得消息在内存和磁盘之间更合理地分配，以支持更大的消息。对于网络传输，可以优化网络设置，如增大 MTU 值（在允许的情况下），减少消息分片传输的开销。同时，合理设置消息队列的缓冲区大小等参数，也有助于提高对大消息的处理能力。

改进拆分策略

  1. 动态调整拆分大小：基于固定大小拆分策略，可以根据消息的实际大小动态调整拆分大小，以减少空间浪费。例如，可以先对消息进行初步分析，根据消息大小的大致范围选择合适的拆分大小。如果消息大小在 100KB - 200KB 之间，选择 100KB 作为拆分大小；如果消息大小在 200KB - 300KB 之间，选择 150KB 作为拆分大小。这样可以在一定程度上提高拆分的效率和资源利用率。
  2. 混合拆分策略：结合多种拆分策略，根据消息的特点灵活选择。对于一些既包含结构化数据又有业务逻辑区分的消息，可以先按照业务逻辑进行初步拆分，然后对拆分后的每个部分再根据数据结构或固定大小进行进一步拆分。这样可以充分发挥不同拆分策略的优势，提高拆分的合理性和适应性。

提高重组可靠性

  1. 增加校验机制：在消息拆分和传输过程中，为每个小消息增加校验和或哈希值等信息。消费者在接收到小消息后，通过验证校验和或哈希值来确保消息的完整性和准确性。如果校验失败，则可以请求重新发送该小消息，从而提高重组的可靠性。
  2. 使用事务机制：在一些支持事务的消息队列中，如 RabbitMQ 可以通过事务机制保证消息的原子性。在拆分和发送消息时，将所有相关的小消息作为一个事务进行处理，要么全部成功发送，要么全部失败回滚。这样可以避免部分消息发送成功而部分失败导致的重组问题。同时，在消息重组过程中，也可以利用事务机制来确保重组操作的完整性。

消息大小限制与拆分策略的实际应用场景

大数据处理场景

  1. 日志收集与分析：在大规模系统中，日志数据量巨大。例如，一个大型电商网站每天可能产生数 TB 的日志数据。这些日志消息需要通过消息队列收集并传输到数据分析平台。由于日志消息通常较大且数量众多，消息队列的消息大小限制就成为一个关键问题。此时，可以采用基于固定大小拆分策略，将大的日志文件拆分成多个小消息进行传输。在数据分析平台端，通过基于消息编号重组的方式将这些小消息还原为原始的日志数据，然后进行分析。
  2. 数据仓库加载：企业的数据仓库需要从各个数据源加载数据。当数据源产生的数据量较大时，如数据库中的大表数据抽取，消息队列在传输这些数据时可能会遇到消息大小限制。可以根据数据结构进行拆分，例如将数据库表中的数据按照行进行拆分，每个小消息包含一定数量的行数据。在数据仓库端，根据数据结构将这些小消息重组为完整的数据表，实现数据的加载。

分布式系统场景

  1. 微服务间通信：在微服务架构中，各个微服务之间通过消息队列进行通信。例如，一个订单微服务在处理复杂订单时，订单消息可能包含商品信息、用户信息、配送信息等大量数据，消息大小可能超出消息队列的限制。此时，可以基于业务逻辑进行拆分，将商品信息发送给商品微服务，用户信息发送给用户微服务，配送信息发送给配送微服务。这些微服务在处理完各自的业务后，再将结果通过消息队列反馈给订单微服务进行重组和最终处理。
  2. 分布式任务调度：在分布式任务调度系统中，一个任务可能包含大量的参数和数据。当将任务通过消息队列发送给各个执行节点时，可能会遇到消息大小问题。可以采用基于固定大小拆分策略，将任务数据拆分成多个小消息发送给执行节点。执行节点在接收到所有相关小消息后，通过基于消息编号重组的方式还原任务数据并执行任务。

物联网场景

  1. 设备数据上传：大量的物联网设备会不断产生数据并上传到云端。例如，智能电表、水表等设备，它们上传的数据可能包含一段时间内的测量数据、设备状态等信息，消息大小可能较大。在这种情况下，可以根据数据结构进行拆分，比如将一段时间内的数据按照固定时间间隔拆分成多个小消息。云端在接收到这些小消息后，根据数据结构进行重组，从而获取完整的设备数据进行分析和处理。
  2. 设备控制指令下发：当云端向物联网设备下发控制指令时，指令消息可能包含复杂的配置信息和操作参数，消息大小也可能超出限制。可以基于业务逻辑进行拆分，将不同类型的配置信息分别拆分成小消息发送给设备。设备在接收到这些小消息后，根据业务逻辑进行重组，解析并执行控制指令。