Standard Message Types

RocketMQ supports three primary message sending modes, each balancing reliability and performance differently.

Synchronous Sending: The producer blocks until it receives an acknowledgment from the broker. This ensures high reliability but incurs latency.

Asynchronous Sending: The producer sends messages without waiting for acknowledgments, using a callback to handle results. This improves throughput while maintaining reasonable reliability.

One-way Sending: The producer dispatches messages without expecting any acknowledgment. This maximizes throughput but offers no delivery guarantee.

Producer Examples

Synchronous:

public class SyncProducerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("producer_group");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.setRetryTimesWhenSendFailed(3);
        sender.setSendMsgTimeout(5000);
        sender.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message("exampleTopic", "exampleTag", ("Hello-" + i).getBytes());
            msg.setKeys("msgKey-" + i);
            SendResult result = sender.send(msg);
            System.out.println(result);
        }
        sender.shutdown();
    }
}

Asynchronous:

public class AsyncProducerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("producer_group");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.setRetryTimesWhenSendAsyncFailed(0);
        sender.setDefaultTopicQueueNums(2);
        sender.start();

        for (int i = 0; i < 100; i++) {
            Message msg = new Message("asyncTopic", "asyncTag", ("Async-" + i).getBytes());
            sender.send(msg, new SendCallback() {
                @Override
                public void onSuccess(SendResult result) {
                    System.out.println("Sent: " + result);
                }
                @Override
                public void onException(Throwable ex) {
                    ex.printStackTrace();
                }
            });
        }
        TimeUnit.SECONDS.sleep(3);
        sender.shutdown();
    }
}

One-way:

public class OnewayProducerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("producer_group");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.start();

        for (int i = 0; i < 10; i++) {
            Message msg = new Message("onewayTopic", "onewayTag", ("Fire-and-forget-" + i).getBytes());
            sender.sendOneway(msg);
        }
        sender.shutdown();
        System.out.println("Producer stopped");
    }
}

Consumer:

public class BasicConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer receiver = new DefaultMQPushConsumer("consumer_group");
        receiver.setNamesrvAddr("rocketmqOS:9876");
        receiver.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        receiver.subscribe("exampleTopic", "*");

        receiver.registerMessageListener((msgs, context) -> {
            msgs.forEach(System.out::println);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        receiver.start();
        System.out.println("Consumer running");
    }
}

Ordered Messaging

Ordered messaging ensures FIFO consumption per partition or globally.

• Global Order: Achieved by restricting a topic to a single queue.
• Partition Order: Messages with the same sharding key are routed to the same queue using a custom MessageQueueSelector.

public class OrderedProducerExample {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("ordered_producer");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.start();

        for (int i = 0; i < 100; i++) {
            Integer orderId = i;
            Message msg = new Message("orderedTopic", "orderedTag", ("OrderMsg-" + i).getBytes());
            sender.send(msg, (mqs, msgObj, arg) -> {
                Integer id = (Integer) arg;
                int idx = id % mqs.size();
                return mqs.get(idx);
            }, orderId);
        }
        sender.shutdown();
    }
}

Delayed Messages

Delayed messages become visible to consumers only after a specified time. RocketMQ uses predefined delay levels (e.g., level 3 = 10 seconds).

Producer:

public class DelayedProducer {
    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("delay_producer");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.start();

        for (int i = 0; i < 10; i++) {
            Message msg = new Message("delayTopic", "delayTag", ("Delayed-" + i).getBytes());
            msg.setDelayTimeLevel(3); // 10-second delay
            sender.send(msg);
            System.out.println(new SimpleDateFormat("mm:ss").format(new Date()) + ", Sent");
        }
        sender.shutdown();
    }
}

Consumer:

public class DelayedConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer receiver = new DefaultMQPushConsumer("delay_consumer");
        receiver.setNamesrvAddr("rocketmqOS:9876");
        receiver.subscribe("delayTopic", "*");
        receiver.registerMessageListener((msgs, ctx) -> {
            msgs.forEach(msg -> System.out.println(new SimpleDateFormat("mm:ss").format(new Date()) + ", Consumed"));
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        receiver.start();
    }
}

Transactional Messages

Transactional messages implement a two-phase commit protocol with optional status checks.

Transaction Listener:

public class BankTransferListener implements TransactionListener {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        String tag = msg.getTags();
        if ("SUCCESS".equals(tag)) return LocalTransactionState.COMMIT_MESSAGE;
        if ("FAILURE".equals(tag)) return LocalTransactionState.ROLLBACK_MESSAGE;
        return LocalTransactionState.UNKNOW;
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // Simulate checking database state
        return LocalTransactionState.COMMIT_MESSAGE;
    }
}

Transactional Producer:

public class TransactionalProducer {
    public static void main(String[] args) throws Exception {
        TransactionMQProducer sender = new TransactionMQProducer("txn_producer");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.setExecutorService(Executors.newFixedThreadPool(3));
        sender.setTransactionListener(new BankTransferListener());
        sender.start();

        String[] outcomes = {"SUCCESS", "FAILURE", "UNCERTAIN"};
        for (int i = 0; i < 3; i++) {
            Message msg = new Message("transferTopic", outcomes[i], ("TxnMsg-" + i).getBytes());
            sender.sendMessageInTransaction(msg, null);
        }
        sender.shutdown();
    }
}

Batch Messaging

Batching improves throughput by sending multiple messages in one request, subject to constraints:

• Same topic and message properties
• Total size ≤ 4 MB (configurable)
• Not applicable to delayed or transactional messages

Batch Producer with Splitting:

public class BatchProducerWithSplitter {
    private static final int MAX_SIZE = 4 * 1024 * 1024;

    public static void main(String[] args) throws Exception {
        DefaultMQProducer sender = new DefaultMQProducer("batch_producer");
        sender.setNamesrvAddr("rocketmqOS:9876");
        sender.start();

        List<Message> messages = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            messages.add(new Message("batchTopic", "batchTag", ("Batch-" + i).getBytes()));
        }

        Iterator<List<Message>> batches = new MessageListSplitter(messages, MAX_SIZE);
        while (batches.hasNext()) {
            sender.send(batches.next());
        }
        sender.shutdown();
    }

    static class MessageListSplitter implements Iterator<List<Message>> {
        private final List<Message> allMessages;
        private final int maxSize;
        private int currentIndex = 0;

        MessageListSplitter(List<Message> messages, int limit) {
            this.allMessages = messages;
            this.maxSize = limit;
        }

        @Override
        public boolean hasNext() {
            return currentIndex < allMessages.size();
        }

        @Override
        public List<Message> next() {
            int start = currentIndex;
            int currentSize = 0;
            while (currentIndex < allMessages.size()) {
                Message msg = allMessages.get(currentIndex);
                int msgSize = estimateSize(msg);
                if (msgSize > maxSize) {
                    if (currentIndex == start) currentIndex++;
                    break;
                }
                if (currentSize + msgSize > maxSize) break;
                currentSize += msgSize;
                currentIndex++;
            }
            return allMessages.subList(start, currentIndex);
        }

        private int estimateSize(Message msg) {
            int size = msg.getTopic().length() + msg.getBody().length + 20;
            for (Map.Entry<String, String> prop : msg.getProperties().entrySet()) {
                size += prop.getKey().length() + prop.getValue().length();
            }
            return size;
        }
    }
}

Batch Consumer:

public class BatchConsumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer receiver = new DefaultMQPushConsumer("batch_consumer");
        receiver.setNamesrvAddr("rocketmqOS:9876");
        receiver.subscribe("batchTopic", "*");
        receiver.setConsumeMessageBatchMaxSize(10);
        receiver.setPullBatchSize(40);

        receiver.registerMessageListener((msgs, ctx) -> {
            msgs.forEach(System.out::println);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });

        receiver.start();
    }
}

Message Filtering

Consumers can filter messages using tags or SQL expressions.

Tag Filtering:

// Consumer subscribes to multiple tags
consumer.subscribe("topic", "tagA || tagB");

SQL Filtering (requires enablePropertyFilter=true in broker config):

// Producer adds user properties
msg.putUserProperty("age", String.valueOf(i));

// Consumer uses SQL expression
consumer.subscribe("topic", MessageSelector.bySql("age BETWEEN 0 AND 6"));

Retry Mechanisms

Sending Retries:

• Synchronous: Retries on different brokers (default 2 times)
• Asynchronous: Retries only on the same broker
• One-way: No retries

Consumption Retries:

• Ordered messages: Infinite retries with fixed interval
• Unordered messages (cluster mode): Exponential backoff up to 16 retries by default

Configure retry behavior:

// For ordered messages
consumer.setSuspendCurrentQueueTimeMillis(100);

// For unordered messages
consumer.setMaxReconsumeTimes(10);

Failed messages beyond retry limits are moved to a dead-letter queue (DLQ) named %DLQ%<consumerGroup>.

Dead-Letter Queues

DLQs store messages that could not be consumed after maximum retries. Key properties:

• One DLQ per consumer group
• Messages expire after 3 days
• Requires manual inspection and reprocessing