大消息处理策略

RabbitMQ对单条消息大小存在实际限制，超大消息需通过分片传输、接收端重组的方式可靠投递。

定义

大消息指超过RabbitMQ推荐大小（建议单条不超过128MB）或网络MTU限制的消息体。处理策略包括：消息分片（Chunking）、分片标识、接收端重组、超时清理。

消息分片模型

原始大消息
    |
    v
+------------------+
| 分片1 | 分片2 | ... | 分片N |
+------------------+
    |         |          |
    v         v          v
  独立发布到RabbitMQ
    |
    v
接收端按group_id收集全部分片后重组

分片消息结构

import java.util.UUID;

public class MessageChunk {
    private String groupId;      // 原始消息唯一标识
    private int chunkIndex;      // 分片序号（从0开始）
    private int totalChunks;     // 总分片数
    private byte[] data;         // 分片数据
    
    public MessageChunk(String groupId, int chunkIndex, int totalChunks, byte[] data) {
        this.groupId = groupId;
        this.chunkIndex = chunkIndex;
        this.totalChunks = totalChunks;
        this.data = data;
    }
    
    // getter/setter省略
    public String getGroupId() { return groupId; }
    public int getChunkIndex() { return chunkIndex; }
    public int getTotalChunks() { return totalChunks; }
    public byte[] getData() { return data; }
}

发送端：消息分片

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.fasterxml.jackson.databind.ObjectMapper;

public class ChunkingProducer {
    private static final String QUEUE_NAME = "chunk_queue";
    private static final int CHUNK_SIZE = 1024 * 1024; // 1MB/分片
    private static final ObjectMapper MAPPER = new ObjectMapper();
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            // 模拟大消息数据（如大文件、大JSON）
            byte[] largeMessage = generateLargePayload();
            
            // 执行分片并发布
            sendInChunks(channel, largeMessage);
        }
    }
    
    private static byte[] generateLargePayload() {
        // 模拟5MB数据
        byte[] payload = new byte[5 * 1024 * 1024];
        for (int i = 0; i < payload.length; i++) {
            payload[i] = (byte) (i % 256);
        }
        return payload;
    }
    
    private static void sendInChunks(Channel channel, byte[] data) throws Exception {
        String groupId = UUID.randomUUID().toString();
        int totalChunks = (int) Math.ceil((double) data.length / CHUNK_SIZE);
        
        for (int i = 0; i < totalChunks; i++) {
            int start = i * CHUNK_SIZE;
            int end = Math.min(start + CHUNK_SIZE, data.length);
            byte[] chunkData = new byte[end - start];
            System.arraycopy(data, start, chunkData, 0, end - start);
            
            MessageChunk chunk = new MessageChunk(groupId, i, totalChunks, chunkData);
            byte[] body = MAPPER.writeValueAsBytes(chunk);
            
            BasicProperties props = new BasicProperties.Builder()
                    .contentType("application/json")
                    .headers(java.util.Map.of(
                            "x-group-id", groupId,
                            "x-chunk-index", i,
                            "x-total-chunks", totalChunks
                    ))
                    .build();
            
            channel.basicPublish("", QUEUE_NAME, props, body);
            System.out.println("已发送分片: " + (i + 1) + "/" + totalChunks);
        }
        
        System.out.println("大消息分片完成, groupId: " + groupId);
    }
}

接收端：分片重组

import com.rabbitmq.client.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.util.*;
import java.util.concurrent.*;

public class ChunkingConsumer {
    private static final String QUEUE_NAME = "chunk_queue";
    private static final ObjectMapper MAPPER = new ObjectMapper();
    
    // 存储待重组的分片: groupId -> (chunkIndex -> chunk)
    private static final ConcurrentHashMap<String, ConcurrentHashMap<Integer, MessageChunk>> pendingChunks = 
            new ConcurrentHashMap<>();
    
    // 分片超时清理（5分钟未收齐则丢弃）
    private static final ScheduledExecutorService CLEANER = 
            Executors.newScheduledThreadPool(1);
    
    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                try {
                    byte[] body = delivery.getBody();
                    MessageChunk chunk = MAPPER.readValue(body, MessageChunk.class);
                    
                    String groupId = chunk.getGroupId();
                    pendingChunks.computeIfAbsent(groupId, k -> new ConcurrentHashMap<>())
                            .put(chunk.getChunkIndex(), chunk);
                    
                    System.out.println("收到分片: " + (chunk.getChunkIndex() + 1) + 
                            "/" + chunk.getTotalChunks() + ", groupId: " + groupId);
                    
                    // 检查是否收齐
                    ConcurrentHashMap<Integer, MessageChunk> group = pendingChunks.get(groupId);
                    if (group != null && group.size() == chunk.getTotalChunks()) {
                        reassembleMessage(groupId, group);
                        pendingChunks.remove(groupId);
                    }
                    
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                } catch (Exception e) {
                    System.err.println("分片处理失败: " + e.getMessage());
                    channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
                }
            };
            
            channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
        }
    }
    
    private static void reassembleMessage(String groupId, ConcurrentHashMap<Integer, MessageChunk> chunks) {
        // 计算总大小
        int totalSize = chunks.values().stream().mapToInt(c -> c.getData().length).sum();
        byte[] reassembled = new byte[totalSize];
        int offset = 0;
        
        // 按chunkIndex顺序重组
        for (int i = 0; i < chunks.size(); i++) {
            MessageChunk chunk = chunks.get(i);
            System.arraycopy(chunk.getData(), 0, reassembled, offset, chunk.getData().length);
            offset += chunk.getData().length;
        }
        
        System.out.println("消息重组完成, groupId: " + groupId + ", 总大小: " + reassembled.length + " bytes");
        // 此处处理完整消息...
    }
}

注意事项

分片大小（CHUNK_SIZE）建议设置在64KB~1MB之间，过小增加网络开销，过大失去分片意义。

必须保证同一groupId的所有分片路由到同一队列，否则重组失败。

接收端需实现超时清理机制，避免因丢失分片导致内存泄漏。

分片传输不具备原子性，部分分片到达期间消费者可能看到不完整数据，需在业务层处理。

对于极端大消息，考虑直接使用对象存储（如S3/OSS），RabbitMQ仅传输引用URL。

替代方案对比

要点总结

• 超大消息需分片传输，规避RabbitMQ单条消息大小限制
• 发送端按固定大小切分，附加groupId、chunkIndex、totalChunks标识
• 接收端收集同一groupId的所有分片后按序重组
• 分片大小建议64KB~1MB，平衡网络开销与分片效果
• 必须实现超时清理，防止丢失分片导致内存泄漏
• 极端大消息建议使用对象存储+URL引用方案