1
# 查看存储结构
2
ls /root/store/commitlog/     # CommitLog 文件（1GB 每个）
3
ls /root/store/consumequeue/  # ConsumeQueue 目录
4
ls /root/store/index/         # IndexFile
5

6
# 查看消息
7
./mqadmin queryMsgById -i <msgId>
8
./mqadmin queryMsgByKey -t TopicOrder -k orderKey123
9
./mqadmin queryMsgByOffset -t TopicOrder -b broker-a -i 0 -o 1000

1
// 1. 创建事务消息生产者
2
TransactionMQProducer producer = new TransactionMQProducer("order-producer-group");
3
producer.setNamesrvAddr("localhost:9876");
4

5
// 2. 设置事务监听器
6
producer.setTransactionListener(new TransactionListener() {
7
    @Override
8
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
9
        try {
10
            // 执行本地事务：创建订单
11
            Order order = JSON.parseObject(msg.getBody(), Order.class);
12
            orderService.createOrder(order);
13
            return LocalTransactionState.COMMIT_MESSAGE;
14
        } catch (Exception e) {
15
            return LocalTransactionState.ROLLBACK_MESSAGE;
16
        }
17
    }
18

19
    @Override
20
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
21
        // 回查：检查本地事务状态
22
        String orderId = msg.getUserProperty("orderId");
23
        Order order = orderService.getOrder(orderId);
24
        if (order != null && order.getStatus() != OrderStatus.CANCELLED) {
25
            return LocalTransactionState.COMMIT_MESSAGE;
26
        }
27
        return LocalTransactionState.ROLLBACK_MESSAGE;
28
    }
29
});
30

31
producer.start();
32

33
// 3. 发送事务消息
34
Message msg = new Message("TopicOrder", "TagA", orderJson.getBytes());
35
msg.putUserProperty("orderId", orderId);
36
producer.sendMessageInTransaction(msg, null);

1
// 发送延迟消息
2
DefaultMQProducer producer = new DefaultMQProducer("delay-producer-group");
3
producer.setNamesrvAddr("localhost:9876");
4
producer.start();
5

6
Message msg = new Message("TopicOrder", "TagA", orderJson.getBytes());
7
msg.setDelayTimeLevel(3);  // 10 秒后消费
8

9
SendResult result = producer.send(msg);
10
System.out.println("Send result: " + result.getSendStatus());

1
// RocketMQ 5.x 任意延迟
2
Message msg = new Message("TopicOrder", "TagA", orderJson.getBytes());
3
msg.setDelayTimeSec(600);    // 延迟 600 秒（10 分钟）
4
// 或
5
msg.setDelayTimeMs(600000L); // 延迟 600000 毫秒
6

7
SendResult result = producer.send(msg);

1
// 场景：订单超时自动取消
2
// 1. 创建订单时发送延迟消息
3
public void createOrder(Order order) {
4
    // 保存订单到数据库
5
    orderMapper.insert(order);
6

7
    // 发送 30 分钟延迟消息
8
    Message msg = new Message("TopicOrderTimeout", order.getId().getBytes());
9
    msg.setDelayTimeLevel(16);  // 30 分钟
10
    producer.send(msg);
11
}
12

13
// 2. 消费超时消息
14
public void onOrderTimeout(MessageExt msg) {
15
    String orderId = new String(msg.getBody());
16
    Order order = orderMapper.selectById(orderId);
17

18
    if (order != null && order.getStatus() == OrderStatus.UNPAID) {
19
        // 订单未支付，自动取消
20
        order.setStatus(OrderStatus.CANCELLED);
21
        orderMapper.updateById(order);
22
        // 释放库存
23
        inventoryService.release(order.getProductId(), order.getQuantity());
24
    }
25
}

1
// 发送顺序消息：使用 MessageQueueSelector
2
DefaultMQProducer producer = new DefaultMQProducer("order-producer-group");
3
producer.start();
4

5
// 按 orderId 选择队列，保证同一订单的消息进入同一队列
6
SendResult result = producer.send(msg, new MessageQueueSelector() {
7
    @Override
8
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
9
        Long orderId = (Long) arg;
10
        int index = (int) (orderId % mqs.size());
11
        return mqs.get(index);
12
    }
13
}, orderId);
14

15
// 顺序消费：使用 MessageListenerOrderly
16
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("order-consumer-group");
17
consumer.subscribe("TopicOrder", "*");
18

19
consumer.registerMessageListener(new MessageListenerOrderly() {
20
    @Override
21
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
22
        for (MessageExt msg : msgs) {
23
            processOrder(msg);
24
        }
25
        return ConsumeOrderlyStatus.SUCCESS;
26
    }
27
});
28
consumer.start();

1
// Producer：设置 Tag
2
Message msg = new Message("TopicOrder", "TagCreate", orderCreate.getBytes());
3
Message msg2 = new Message("TopicOrder", "TagPay", orderPay.getBytes());
4

5
// Consumer：按 Tag 订阅
6
// 方式一：订阅单个 Tag
7
consumer.subscribe("TopicOrder", "TagCreate");
8

9
// 方式二：订阅多个 Tag（|| 分隔）
10
consumer.subscribe("TopicOrder", "TagCreate||TagPay");
11

12
// 方式三：订阅所有 Tag
13
consumer.subscribe("TopicOrder", "*");

1
// Producer：设置属性
2
Message msg = new Message("TopicOrder", "TagA", orderJson.getBytes());
3
msg.putUserProperty("amount", "1500");
4
msg.putUserProperty("region", "east");
5
msg.putUserProperty("vip", "true");
6

7
// Consumer：SQL92 过滤
8
consumer.subscribe("TopicOrder",
9
    MessageSelector.bySql("amount > 1000 AND region = 'east'"));
10

11
// 其他 SQL92 示例
12
MessageSelector.bySql("vip = 'true'");
13
MessageSelector.bySql("amount BETWEEN 100 AND 1000");
14
MessageSelector.bySql("region IN ('east', 'south')");
15
MessageSelector.bySql("amount > 500 OR vip = 'true'");

1
# 开启 SQL92 过滤（Broker 配置）
2
enablePropertyFilter=true

1
DefaultMQProducer producer = new DefaultMQProducer("reliable-producer-group");
2
producer.setRetryTimesWhenSendFailed(3);          // 同步发送重试次数
3
producer.setRetryTimesWhenSendAsyncFailed(3);     // 异步发送重试次数
4
producer.setSendLatencyFaultEnable(true);          // 开启故障延迟
5

6
// 同步发送（最可靠）
7
SendResult result = producer.send(msg);
8
if (result.getSendStatus() != SendStatus.SEND_OK) {
9
    // 处理发送失败
10
    log.error("Send failed: {}", result.getSendStatus());
11
}

1
# Broker 配置（高可靠模式）
2
flushDiskType=SYNC_FLUSH        # 同步刷盘
3
brokerRole=SYNC_MASTER          # 同步复制
4
sendMessageThreadPoolNums=64     # 发送线程池
5
flushCommitLogTimed=0           # 实时刷盘

1
// 消费重试机制
2
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("reliable-consumer-group");
3

4
// 并发消费：失败后自动重试
5
consumer.registerMessageListener(new MessageListenerConcurrently() {
6
    @Override
7
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
8
        try {
9
            for (MessageExt msg : msgs) {
10
                processMessage(msg);
11
            }
12
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
13
        } catch (Exception e) {
14
            // 返回 RECONSUME_LATER，消息会进入重试队列
15
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
16
        }
17
    }
18
});
19

20
// 设置最大重试次数
21
consumer.setMaxReconsumeTimes(16);  // 默认 16 次