RocketMQ 面试题

Topic (主题)
  ├── Queue 0 ──→ Consumer Instance 1
  ├── Queue 1 ──→ Consumer Instance 2
  ├── Queue 2 ──→ Consumer Instance 3
  └── Queue 3 ──→ Consumer Instance 4

// 集群消费（默认）
consumer.setMessageModel(MessageModel.CLUSTERING);

// 广播消费
consumer.setMessageModel(MessageModel.BROADCASTING);

// 同步发送（推荐）
SendResult result = producer.send(msg);

// 异步发送 + 回调
producer.send(msg, new SendCallback() {
    @Override
    public void onSuccess(SendResult sendResult) {}
    @Override
    public void onException(Throwable e) {
        // 重试或记录日志
    }
});

# 同步刷盘（更可靠）
flushDiskType=SYNC_FLUSH

# 同步复制（更可靠）
brokerRole=SYNC_MASTER

consumer.registerMessageListener((msgs, context) -> {
    try {
        // 处理消息
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (Exception e) {
        // 返回 RECONSUME_LATER 触发重试
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});

public boolean consume(Message msg) {
    String msgId = msg.getMsgId();
    
    // 分布式锁 + 去重
    if (redis.setNx("consumed:" + msgId, "1", 7*24*3600)) {
        try {
            process(msg);
            return true;
        } catch (Exception e) {
            redis.del("consumed:" + msgId);
            throw e;
        }
    }
    return true;  // 已消费，直接返回成功
}

// 发送端：相同订单号进同一队列
producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
        Long orderId = (Long) arg;
        int index = (int) (orderId % mqs.size());
        return mqs.get(index);
    }
}, orderId);

// 消费端：使用顺序消费监听器
consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
        // 顺序处理
        return ConsumeOrderlyStatus.SUCCESS;
    }
});

msg.setDelayTimeLevel(3);  // 10s 后投递

// 延迟级别（18 个）:
// 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h

// 设置投递时间戳
msg.setDeliverTimeMs(System.currentTimeMillis() + 60000);  // 1 分钟后

Producer ───────────────────── Broker ─────────────────── Consumer
    │                             │                           │
    │──① 发送半消息(Prepare)────→│                           │
    │←──② 返回发送结果──────────│                           │
    │                             │                           │
    │──③ 执行本地事务             │                           │
    │                             │                           │
    │──④ Commit/Rollback────────→│                           │
    │                             │──⑤ 消息可消费─────────→ │
    │                             │                           │
    │←──⑥ 回查本地事务状态(如需)──│                           │

TransactionMQProducer producer = new TransactionMQProducer("group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务
            doLocalTransaction();
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 回查本地事务状态
        if (checkTransactionSuccess(msg)) {
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.ROLLBACK_MESSAGE;
    }
});

// 发送事务消息
producer.sendMessageInTransaction(msg, null);

# Broker 配置
flushDiskType=SYNC_FLUSH   # 同步刷盘
flushDiskType=ASYNC_FLUSH  # 异步刷盘（默认）

// 增加消费线程
consumer.setConsumeThreadMax(64);
consumer.setConsumeThreadMin(32);

// 批量消费
consumer.setConsumeMessageBatchMaxSize(10);

# Master 配置
brokerRole=ASYNC_MASTER   # 异步复制
brokerRole=SYNC_MASTER    # 同步复制

# Slave 配置
brokerRole=SLAVE

// Tag 过滤
consumer.subscribe("Topic", "TagA || TagB");

// SQL92 过滤
consumer.subscribe("Topic", MessageSelector.bySql("age > 18"));

// 开启消息轨迹
DefaultMQProducer producer = new DefaultMQProducer("group", true);

// 手动提交（推荐）
consumer.setAutoCommit(false);
consumer.commitSync();