Java 面试 200 题

主题

九、RocketMQ（10 题） ​

9.1 ★★★★RocketMQ 的整体架构是怎样的？NameServer、Broker、Producer、Consumer 各自的角色和交互流程是什么？ ​

📖 点击查看答案

整体架构：RocketMQ 采用分布式架构，由 NameServer、Broker、Producer、Consumer 四大组件构成，Broker 分 Master/Slave，通过 Topic 与队列组织消息。

  • NameServer：轻量级注册中心，无状态，可集群部署互不通信；维护 Broker 路由信息（Topic→Queue→Broker 映射）。
  • Broker：消息存储与转发服务器，Master 支持读写，Slave 仅支持读；定时（默认 30s）向所有 NameServer 发送心跳注册。
  • Producer：消息生产者，启动时从 NameServer 拉取路由信息，与 Broker 建立长连接发送消息。
  • Consumer：消息消费者，从 NameServer 获取路由后与 Broker 建立长连接，拉取消息并消费。

交互流程：

  1. Broker 启动 → 向所有 NameServer 注册心跳（IP、Topic、Queue 信息）。
  2. Producer/Consumer 启动 → 从任意一台 NameServer 拉取路由信息，本地缓存（默认 30s 更新）。
  3. Producer 发送消息 → 根据 Topic 找到队列，选择 Broker（轮询/指定），发送。
  4. Consumer 订阅 Topic → 根据负载均衡策略拉取消息，消费完成后返回 ACK。
  5. NameServer 与 Broker 心跳超时（默认 120s）→ 剔除该 Broker 节点，通知客户端更新路由。
java
// Producer 示例
DefaultMQProducer producer = new DefaultMQProducer("group_name");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.start();
Message msg = new Message("TopicA", "TagA", "hello".getBytes());
SendResult result = producer.send(msg);

关键点：NameServer 之间无通信，依赖 Broker 主动注册多台；Broker 主从同步支持同步/异步复制；Producer 选择队列时支持故障隔离（sendLatencyFaultEnable）。

9.2 ★★RocketMQ 如何保证消息不丢失？分别从 Producer、Broker、Consumer 三个环节说明。 ​

📖 点击查看答案

Producer 端：

  • 同步发送 + 重试：producer.send(msg)，失败自动重试（默认 2 次）。
  • 使用 sendSync 保证 Broker 收到 ACK 才认为成功。
  • 异步发送需在回调中处理失败逻辑并落库重发。

Broker 端：

  • 同步刷盘：flushDiskType=SYNC_FLUSH，消息写入磁盘后才返回 ACK（性能低）。
  • 同步复制：brokerRole=SYNC_MASTER，Master 等 Slave 同步完成后再返回。
  • 开启 translog 持久化，防止宕机丢数据。

Consumer 端：

  • 手动 ACK：消费成功后 consumer.consumeSuccess()，失败时返回 RECONSUME_LATER。
  • 关闭自动提交 offset，使用 MessageListenerConcurrently 中 status 控制。
  • 幂等消费，避免重试导致重复消费问题。
java
// Broker 配置
brokerRole=SYNC_MASTER
flushDiskType=SYNC_FLUSH

// Consumer 手动 ACK
consumer.registerMessageListener((msgs, context) -> {
    try {
        // 业务处理
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    } catch (Exception e) {
        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
    }
});

权衡：同步刷盘 + 同步复制 = 最高可靠性但性能最低，金融场景使用；高吞吐场景可用异步刷盘 + 异步复制 + 消息重试兜底。

9.3 ★★如何保证消息发送和消费的幂等性？消费端如何实现幂等？ ​

📖 点击查看答案

消息发送幂等：

  • RocketMQ 4.x+ 支持消息轨迹与 MessageQueueSelector，可基于业务 key 路由到固定队列。
  • Producer 发送失败重试可能产生重复，需消费端幂等兜底。
  • 业务层生成全局唯一 ID（雪花算法/UUID）放入 keys 字段。

消费端幂等实现方案：

  1. 数据库唯一索引：业务表对 msgKey 建唯一索引，重复插入报错即跳过。
  2. Redis SETNX：消费前 SET msgKey NX EX 86400，成功则消费。
  3. 状态机校验：订单状态机（待支付→已支付），重复消费时状态已变更直接返回成功。
  4. 去重表：单独建表记录已消费的 msgId，消费前查询。
java
// Redis 去重示例
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
    for (MessageExt msg : msgs) {
        String key = msg.getKeys();
        Boolean absent = redis.opsForValue().setIfAbsent("mq:" + key, "1", 24, TimeUnit.HOURS);
        if (Boolean.FALSE.equals(absent)) {
            continue; // 已消费，跳过
        }
        // 业务处理
        doBusiness(msg);
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}

关键点：RocketMQ 不保证Exactly-Once，需业务侧实现；幂等 key 优先使用业务唯一 ID，避免依赖 msgId（重试时可能不同）。

9.4 ★★★RocketMQ 的消费模式有哪些？集群消费和广播消费的区别是什么？ ​

📖 点击查看答案

两种消费模式：

  • 集群消费（CLUSTERING）：默认模式，同一 ConsumerGroup 下多个 Consumer 平分消息，每条消息只被消费一次。
  • 广播消费（BROADCASTING）：同一 ConsumerGroup 下每个 Consumer 都消费全量消息。

核心区别：

维度集群消费广播消费
消息分发队列维度负载均衡每个消费者都收到
offset 存储Broker 端Consumer 本地
失败重试支持（重试队列）不支持（消费失败丢弃）
适用场景订单处理、日志聚合配置广播、本地缓存同步
java
// 集群消费（默认）
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group_name");
consumer.setMessageModel(MessageModel.CLUSTERING);

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

负载均衡机制：

  • 集群模式下，Consumer 启动时根据消费模式（平均分配/一致性哈希/机房优先）分配 Queue。
  • 默认 AllocateMessageQueueAveragely 策略：Q 数 ≥ C 数时每个 C 至少分配 ⌊Q/C⌋ 个。
  • Queue 数与 Consumer 数不匹配时会有不均衡，建议 Queue 数为 Consumer 数的整数倍。

注意：广播模式下 offset 存本地，Broker 重启不影响，但 Consumer 重启会从头消费（除非持久化 offset）。

9.5 ★★RocketMQ 的消费者负载均衡机制是怎样的？推模式和拉模式的区别？ ​

📖 点击查看答案

负载均衡机制：

  • 基于 Queue 维度而非消息维度，Broker 每个 Topic 默认 4 个 ReadQueue/WriteQueue。
  • Consumer 启动后定期（默认 20s）触发 Rebalance，根据策略分配 Queue。
  • 内置分配策略：平均分配、环形分配、一致性哈希、机房优先、一致性哈希。
java
// 自定义分配策略
consumer.setAllocateMessageQueueStrategy(new AllocateMessageQueueAveragely());

推模式（Push）vs 拉模式（Pull）：

维度Push 模式Pull 模式
APIDefaultMQPushConsumerDefaultMQPullConsumer / LitePullConsumer
实时性高（长轮询）取决于拉取间隔
流控内置手动控制
实现本质是 Pull，封装了长轮询用户主动拉取
offset自动管理手动管理

长轮询机制：Push 模式下 Consumer 发送拉取请求，Broker 无消息时挂起请求（默认 15s），期间有消息立即返回，平衡实时性与性能。

java
// Push 模式（推荐）
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group");
consumer.registerMessageListener((msgs, ctx) -> ConsumeConcurrentlyStatus.CONSUME_SUCCESS);

// LitePullConsumer（5.x 推荐）
LitePullConsumer litePull = new LitePullConsumer("group");
litePull.subscribe("TopicA", "*");
List<MessageExt> msgs = litePull.poll(1000);

关键点：Push 模式更易用，但消费速度跟不上会产生积压；Pull 模式灵活可控，适合批量消费场景。

9.6 ★★★★RocketMQ 的事务消息原理是什么？半消息（Half Message）和事务回查机制是如何工作的？ ​

📖 点击查看答案

事务消息原理：采用两阶段提交 + 事务回查机制，保证本地事务与消息发送的最终一致性。

半消息（Half Message）：Producer 先发送一条特殊消息到 Broker，该消息对 Consumer 不可见（存入 RMQ_SYS_TRANS_HALF_TOPIC），等本地事务执行后再决定提交或回滚。

完整流程：

  1. Producer 发送半消息 → Broker 暂存到半消息 Topic，返回发送成功。
  2. Producer 执行本地事务（订单入库等）。
  3. 根据本地事务结果：
    • COMMIT_MESSAGE → Broker 将消息从半消息 Topic 移到真实 Topic，Consumer 可消费。
    • ROLLBACK_MESSAGE → Broker 删除半消息。
    • UNKNOW → Broker 等待事务回查。
  4. 事务回查：Broker 定时（默认 60s）扫描半消息，回调 Producer 的 checkLocalTransaction 方法查询本地事务状态，根据返回结果提交/回滚。
java
TransactionMQProducer producer = new TransactionMQProducer("tx_group");
producer.setTransactionListener(new TransactionListener() {
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        try {
            // 执行本地事务
            return LocalTransactionState.COMMIT_MESSAGE;
        } catch (Exception e) {
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 查询本地事务状态
        return isCommitted(msg) ? LocalTransactionState.COMMIT_MESSAGE : LocalTransactionState.ROLLBACK_MESSAGE;
    }
});
producer.sendMessageInTransaction(msg, null);

回查机制要点：

  • 回查次数默认 15 次，超过后默认回滚（transactionCheckMax）。
  • 回查间隔默认 60s。
  • Producer 必须实现 checkLocalTransaction，需通过业务表状态/事务日志判断本地事务是否成功。

注意：事务消息仅保证 Producer 端的一致性，Consumer 端仍需幂等消费。

9.7 ★RocketMQ 的顺序消息是如何实现的？全局顺序和分区顺序有什么区别？延迟消息如何实现？ ​

📖 点击查看答案

顺序消息实现：

  • 生产端：通过 MessageQueueSelector 将同一业务 key 的消息发送到同一 Queue。
  • 消费端：使用 MessageListenerOrderly 串行消费同一 Queue 的消息（加锁机制）。
java
// 生产端：按 orderId 路由到固定 Queue
producer.send(msg, (mqs, message, arg) -> {
    int index = Math.abs(arg.hashCode()) % mqs.size();
    return mqs.get(index);
}, orderId);

// 消费端：顺序消费
consumer.registerMessageListener((List<MessageExt> msgs, ConsumeOrderlyContext ctx) -> {
    // 同一 Queue 串行执行
    return ConsumeOrderlyStatus.SUCCESS;
});

全局顺序 vs 分区顺序：

  • 全局顺序：Topic 只设 1 个 Queue，所有消息严格有序，吞吐量极低，少用。
  • 分区顺序：多 Queue，同一 key 的消息有序，吞吐量高，推荐使用。

延迟消息实现：

  • RocketMQ 4.x 开源版支持 18 个固定延迟级别（1s/5s/10s/.../2h），不支持任意延迟。
  • 原理：消息发送后存入 SCHEDULE_TOPIC_XXXX（按延迟级别分 Queue），定时任务扫描到期消息后改写 Topic 为目标 Topic。
  • RocketMQ 5.x 支持任意时间延迟（基于时间轮算法）。
java
// 设置延迟级别（4.x）
Message msg = new Message("TopicA", "data".getBytes());
msg.setDelayTimeLevel(3); // 10s 延迟

// 5.x 任意延迟
msg.setDeliverTimeMs(System.currentTimeMillis() + 60000); // 1 分钟后

注意：顺序消费会牺牲并发度，且消费失败会阻塞当前 Queue；延迟消息在 4.x 受限于 18 个级别。

9.8 ★★RocketMQ 的消息重试机制是怎样的？死信队列（DLQ）是什么？如何避免消费重试风暴？ ​

📖 点击查看答案

消息重试机制：

  • 顺序消费：失败后无限重试，MAX_TIME 为 Long.MAX_VALUE，间隔通过 suspendCurrentQueueTimeMillis 控制（默认 1s）。
  • 并发消费：失败后进入重试队列 %RETRY%group，默认重试 16 次，间隔递增（1s、5s、10s、30s、1m、2m、3m、4m、5m、6m、7m、8m、9m、10m、20m、30m、2h、2h）。
  • 重试超过最大次数（默认 16）后进入死信队列。

死信队列（DLQ）：

  • Topic 命名：%DLQ%consumerGroup。
  • 特点：消息不再被自动消费，需人工介入处理。
  • 死信消息保留时间默认 72 小时，过期自动删除。
  • 可通过 DefaultMQPushConsumer 的 viewMessage 查询，或 Dashboard 重发。

避免重试风暴策略：

  1. 设置合理的最大重试次数（maxReconsumeTimes）。
  2. 消费端快速失败：异常时不要长时间阻塞，返回 RECONSUME_LATER。
  3. 死信告警：监听 DLQ Topic，有消息时告警人工处理。
  4. 限流降级：消费速度跟不上时先降级，避免雪崩。
  5. 异常分类处理：业务异常直接 ACK 跳过（记录日志），系统异常才重试。
java
// 控制重试次数
consumer.setMaxReconsumeTimes(5);

// 区分异常
try {
    doBusiness(msg);
} catch (BusinessException e) {
    // 业务异常，不重试
    log.error("biz error", e);
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} catch (Exception e) {
    // 系统异常，重试
    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}

关键点：重试会占用队列资源，重试过多可能导致积压；建议非核心业务失败即丢弃 + 告警。

9.9 ★★★RocketMQ 的消息积压如何处理？有哪些排查和解决思路？如何快速消费积压的消息？ ​

📖 点击查看答案

排查思路：

  1. 查看消费 TPS 与生产 TPS，确认是否消费速度 < 生产速度。
  2. 检查 Consumer 实例数是否足够，是否有节点假死或频繁 GC。
  3. 检查消费逻辑是否有外部依赖瓶颈（DB 慢查询、下游接口超时）。
  4. 检查是否大量重试消息占用资源（重试风暴）。
  5. 查看队列堆积数（consumerLag），通过 Dashboard 或 mqadmin 命令。

解决思路：

  1. 临时扩容 Consumer：增加消费者实例（注意不能超过 Queue 数，超出无效）。
  2. 增加 Queue 数 + 临时 Consumer：扩 Queue 后，临时 Topic 转发到新 Topic（多 Queue），再启多个 Consumer 消费。
  3. 降级非核心消费：关闭非核心业务消费，集中资源消费核心 Topic。
  4. 优化消费逻辑：
    • 批量消费：consumer.setConsumeMessageBatchMaxSize(50)。
    • 异步处理：消息接收后丢线程池异步处理，快速返回 ACK。
    • 减少外部调用：合并 DB 查询、加缓存。
  5. 跳过堆积：若消息有时效性（如行情），可调整 offset 跳过部分消息。
java
// 方案：临时 Topic 转发加速消费
// 1. 原 Consumer 仅做转发
consumer.registerMessageListener((msgs, ctx) -> {
    for (MessageExt msg : msgs) {
        Message newMsg = new Message("TempTopic", msg.getBody());
        newMsg.setKeys(msg.getKeys());
        producer.send(newMsg);  // 转发到多 Queue 的临时 Topic
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
// 2. TempTopic 配置 20 个 Queue，启动 20 个 Consumer 消费

长期治理：

  • 监控告警：堆积超阈值告警（Prometheus + Grafana）。
  • 容量规划：根据峰值 QPS 预留消费能力。
  • 消费端隔离：核心与非核心 Topic 用不同 ConsumerGroup。

9.10 ★★RocketMQ 和 Kafka 的对比？架构设计、性能特点、适用场景的差异？ ​

📖 点击查看答案

架构对比：

维度RocketMQKafka
注册中心NameServer（无状态，AP）Zookeeper（强一致，CP）
存储模型Topic 多 Queue 共享 CommitLogTopic-Partition 独立文件
主从复制支持 Master-Slave（同步/异步）ISR 副本机制
事务消息支持事务消息（半消息+回查）仅支持幂等 Producer + 事务（0.11+）
消费模式集群 + 广播集群（ConsumerGroup）
延迟消息内置支持不支持（需自实现）
死信队列内置不支持

性能特点：

维度RocketMQKafka
单机 TPS10 万级百万级（批量场景）
延迟毫秒级毫秒级
消息大小支持大消息（4MB 默认）适合小消息
堆积能力亿级（CommitLog 顺序写）亿级（依赖 PageCache）
消费模型长轮询Pull

适用场景差异：

  • RocketMQ：金融、电商交易、订单状态流转，对消息可靠性、事务、延迟消息有要求的业务场景。
  • Kafka：大数据流处理、日志采集、用户行为埋点，高吞吐、批量处理的场景。

设计哲学差异：

  • RocketMQ 设计偏向业务消息，提供丰富的消息类型（事务、延迟、顺序、死信）。
  • Kafka 设计偏向数据管道，强调吞吐量与流处理生态（Kafka Streams、Connect）。

选型建议：业务消息选 RocketMQ，大数据/日志场景选 Kafka；中小团队优先 RocketMQ（运维简单，无 ZK 依赖）。

基于 4 份主流面试题库整合精选，共 200 题

仅供学习参考