Message Sharding Example

In distributed high-concurrency systems, as a mainstream message middleware, RocketMQ's performance bottlenecks often concentrate on the throughput ceiling of its message queues. When individual messages exceed acceptable sizes or single write volumes surpass node processing capabilities, direct writes lead to surging disk I/O pressure, increased risks of Out Of Memory (OOM) errors, and even message backlog issues. In such scenarios, introducing a "message sharding" mechanism becomes crucial for resolving these problems. Message sharding is not merely simple data slicing; rather, it is an architectural strategy that decomposes logically large messages into multiple physical small units and distributes them across different Broker nodes or various partitions (Partition) within the same queue. This article uses RocketMQ as a case study to deeply analyze the implementation principles of message sharding, core parameter configurations, and practical application scenarios.
RocketMQ's message sharding is primarily based on its internal message serialization and routing mechanisms. By default, a single message is stored as a complete entity within a specific Partition of a Topic. However, when business scenarios require handling oversized files or high-frequency writes, the system decomposes individual logical messages into multiple "sub-messages" (Sub-message).

This process typically occurs at either the producer end or the consumer end. Taking the producer as an example, upon detecting that the message body length exceeds a preset threshold, the producer automatically truncates long messages into several short segments. Each sub-message carries identical business identifiers (such as Transaction IDs or Order Numbers) but possesses independent MessageQueue affiliations. At the storage layer, these sub-messages are written to different Broker nodes or different Partitions within the same Topic in a distributed manner. This distributed storage approach not only reduces I/O pressure on single nodes but also enhances overall system fault tolerance—provided that other nodes hold data fragments of this business logic, data integrity remains guaranteed even if a node fails.

Key Configuration Parameters and Best Practices

To effectively implement message sharding strategies, reasonable configuration of related parameters is critical. At the RocketMQ Producer end, key control points lie in attributes such as maxMessageSize and batchMaxMessages.

  • Maximum Message Size Limit: This serves as the threshold triggering sharding. It is recommended to adjust the default value (typically 256KB) based on actual business data characteristics. For applications involving large images, JSON big data, or log streams, this threshold can be appropriately lowered to trigger sharding logic earlier.
  • Batch Write Strategy: The batchMaxMessages parameter determines the maximum number of messages sent to a Broker in each batch. A reasonable batch size balances network overhead with memory consumption, avoiding timeouts caused by excessively large single-network requests.
  • Transaction Consistency Assurance: In sharding scenarios, it is essential to ensure strict consistency among all sub-messages corresponding to the same business logic. This is typically achieved through "half-message" mechanisms or distributed transaction protocols. When a Producer sends sub-messages, they are marked as half-messages; only after all fragments have been successfully written will the transaction status be committed, preventing situations where some succeed while others fail.

Typical Application Scenarios and Code Examples

Message sharding is particularly common in scenarios such as e-commerce transactions, log collection, and file transfer. Below is a simplified message sharding logic example based on the Java client, demonstrating how to decompose one large message into multiple sub-messages for sending:

import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.MessageExt;

public class MessageShardingExample {

    private static final int MAX_MESSAGE_SIZE = 1024 * 1024; // Threshold of 1MB
    private static final String TOPIC_NAME = "ORDER_TOPIC";

    public void sendLargeMessage(String orderData) {
        byte[] messageBytes = orderData.getBytes(StandardCharsets.UTF_8);
        int totalSize = messageBytes.length;
        int chunkSize = MAX_MESSAGE_SIZE / 10; // Each fragment approx. 100KB
        
        List<String> subMessages = new ArrayList<>();
        
        // Loop to split into sub-messages
        for (int i = 0; i < totalSize; i += chunkSize) {
            byte[] chunk = Arrays.copyOfRange(messageBytes, i, Math.min(i + chunkSize, totalSize));
            
            MessageExt subMsg = new MessageExt();
            subMsg.setTopic(TOPIC_NAME);
            subMsg.setBody(chunk);
            // Set business identifier for subsequent reassembly or verification
            subMsg.putUserCustomProperty("orderId", "ORD-" + System.currentTimeMillis()); 
            subMsg.putUserCustomProperty("fragmentIndex", String.valueOf(i / chunkSize));
            
            subMessages.add(subMsg.toString());
        }

        // Batch send sharded messages
        for (String msgStr : subMessages) {
            try {
                producer.send(msgStr, new SendCallback() {
                    @Override
                    public void onSuccess(SendResult sendResult) {
                        System.out.println("Message sent successfully: " + sendResult.getMsgIds());
                    }

                    @Override
                    public void onError(Throwable ex) {
                        System.err.println("Send failed: " + ex.getMessage());
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

The above code demonstrates how to cut a large message exceeding the threshold into multiple sub-messages and attach business identifiers and fragment indices to each. In actual production environments, efficiency is further improved by combining this with RocketMQ's batch sending interface (BatchSend), while utilizing retry mechanisms to handle send failures potentially caused by network jitter.

Summary and Outlook

Message sharding represents an important technical practice for building highly available and high-performance RocketMQ applications. By converting heavy loads into small units, it effectively circumvents bottlenecks in single-node storage and I/O, thereby enhancing system scalability. However, sharding also introduces new complexities regarding data reassembly and transaction consistency maintenance. When designing systems, developers should weigh the overhead against the benefits of sharding and dynamically adjust strategies based on the actual size distribution of business data. As RocketMQ versions iterate, its native support for large messages continues to strengthen; however, understanding the underlying logic of sharding remains beneficial for making optimal architectural decisions when facing extreme high-concurrency scenarios.