（衔接我之前的 Kafka 笔记，补充 RocketMQ 对比 + 跨 MQ 实战）
之前写了 Kafka 的核心笔记（聚焦分区、Offset、高效读写等），但实际项目中常遇到 “Kafka 和 RocketMQ 怎么选？”“RocketMQ 原生 Client 和 Spring 集成怎么用？”“RocketMQ 消息转 Kafka 怎么落地？” 这类问题。这篇笔记就把 Kafka 和 RocketMQ 做核心对比，再聚焦RocketMQ→Kafka 转发的两种实现方式（原生 Client/ Spring 集成），同时补充消费组、降级策略的双模式实战。

一、回顾：Kafka 的核心定位

先快速回顾 Kafka 的核心设计（对应我之前的笔记内容）：

• 定位：高吞吐的日志型消息队列，主打大数据流式处理、日志采集；
• 核心设计：Topic+Partition 分区模型、Offset 消费位点管理、顺序写 + 零拷贝实现高效读写；
• 局限：原生不支持事务 / 延迟消息，可靠性配置需手动调整。

二、Kafka vs RocketMQ：核心差异对比

这是选择 MQ 的关键，我们按核心维度逐一对比：

• 研发与生态：Kafka 由 LinkedIn 开源，深度适配大数据生态，和 Flink、Spark 等流式计算框架集成紧密；RocketMQ 是阿里开源的中间件，更适配电商、金融等核心业务场景，能无缝对接阿里系技术栈。
• 核心定位：Kafka 侧重 “日志型”，以超高吞吐为核心优势；RocketMQ 则是 “业务型” 消息队列，在保证高吞吐的同时，更注重消息可靠性和业务适配性。
• 消息模型：Kafka 采用 Topic+Partition 的设计，分区是最小的并行消费单元；RocketMQ 则是 Topic+Queue 的模型，队列作为最小并行单元，负载均衡策略更灵活。
• 消费组管理：两者都支持消费者启动时自动创建消费组；不同的是，RocketMQ 除了自动创建，还支持通过原生 Client 的 Admin 工具编程式配置消费组属性，而 Kafka 的定制化配置需借助 AdminClient。
• 适用场景：Kafka 更适合日志采集、大数据流式处理等对吞吐要求极高的场景；RocketMQ 则更适合电商交易、核心业务异步解耦、跨系统消息同步等对可靠性和业务适配性要求高的场景。

三、实战：RocketMQ 核心操作（分原生 Client/ Spring 集成）

这部分聚焦 RocketMQ→Kafka 转发，同时区分两种实现方式。

3.1 跨 MQ 互发：RocketMQ → Kafka（双模式实现）

场景：RocketMQ 接收业务消息后，转发到 Kafka 供大数据系统消费。

3.1.1 原生 RocketMQ Client 实现

通过原生DefaultMQPushConsumer消费 RocketMQ，搭配 Kafka 原生KafkaProducer发送（无框架依赖）：

// 1. 初始化Kafka原生生产者
Properties kafkaProps = new Properties();
kafkaProps.put("bootstrap.servers", "localhost:9092");
kafkaProps.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
kafkaProps.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(kafkaProps);

// 2. 初始化RocketMQ原生消费者
DefaultMQPushConsumer rocketConsumer = new DefaultMQPushConsumer("rocket2kafka_consumer_group");
rocketConsumer.setNamesrvAddr("localhost:9876");
rocketConsumer.subscribe("rocket_biz_topic", "*"); // 订阅RocketMQ业务主题

// 3. 消费RocketMQ并转发到Kafka
rocketConsumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        try {
            String content = new String(msg.getBody(), "UTF-8");
            // 构建Kafka消息并同步发送（保证可靠性）
            ProducerRecord<String, String> kafkaMsg = new ProducerRecord<>("kafka_data_topic", content);
            kafkaProducer.send(kafkaMsg).get();
            System.out.printf("原生Client转发成功：Rocket消息ID=%s%n", msg.getMsgId());
        } catch (Exception e) {
            System.err.println("转发失败：" + e.getMessage());
            return ConsumeConcurrentlyStatus.RECONSUME_LATER; // 触发RocketMQ重试
        }
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});

// 启动RocketMQ消费者
rocketConsumer.start();

3.1.2 Spring Boot 集成实现

通过@RocketMQMessageListener注解消费 RocketMQ，搭配KafkaTemplate发送（Spring 生态更简洁）：
步骤 1：配置文件（application.yml）

yaml
spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
rocketmq:
  name-server: localhost:9876
  consumer:
    group: rocket2kafka_spring_group

步骤 2：核心代码

@SpringBootApplication
public class Rocket2KafkaSpringApp {
    public static void main(String[] args) {
        SpringApplication.run(Rocket2KafkaSpringApp.class, args);
    }
}

// Spring注解式消费RocketMQ
@Component
@RocketMQMessageListener(topic = "rocket_biz_topic", consumerGroup = "rocket2kafka_spring_group")
public class RocketSpringConsumer implements RocketMQListener<String> {
    // 注入Spring封装的KafkaTemplate
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void onMessage(String content) {
        // 1. 异步发送Kafka消息，返回ListenableFuture（Spring Kafka核心异步API）
        ListenableFuture<SendResult> future = kafkaTemplate.send("kafka_data_topic", content);
        
        // 2. 注册异步回调：分别处理成功/失败场景
        future.addCallback(
            // onSuccess：Kafka发送成功的逻辑
            sendResult -> {
                System.out.printf("Spring版转发成功：内容=%s，Kafka分区=%d，偏移量=%d%n",
                        content,
                        sendResult.getRecordMetadata().partition(),
                        sendResult.getRecordMetadata().offset());
            },
            // onFailure：Kafka发送失败的逻辑
            throwable -> {
                System.err.printf("Spring版转发失败：内容=%s，异常=%s%n", content, throwable.getMessage());
                // 抛出运行时异常，触发RocketMQ的重试机制（核心：失败必抛异常）
                throw new RuntimeException("Kafka消息发送失败，触发RocketMQ重试", throwable);
            }
        );
    }
}

3.1.3 生产环境切换：三大核心风险规避

在新旧转发服务切换（如从原生 Client 切到 Spring 集成版）时，需重点规避以下三类问题，确保无丢失、无重复、无配置冲突：
风险 1：上下线导致消息丢失（解决方案：优雅下线 + 指定消费位点）

• 旧服务下线：必须通过consumer.shutdown()优雅关闭，等待消费线程处理完当前消息、提交 Offset 后再停止，避免未处理的消息丢失；
• 新服务上线：通过consumeFromWhere = CONSUME_FROM_FIRST_OFFSET配置，强制从最早消费位点开始消费，避免切换期间的消息漏消费；
• 兜底保障：在 RocketMQ 控制台延长 Offset 保留时间（默认 72 小时，可调整为 168 小时），防止旧服务下线期间 Offset 被清理。

风险 2：双服务并行导致重复消费（解决方案：幂等处理 + 分组隔离）
若需要灰度切换（新旧服务临时并行运行），需通过 “分组隔离 + 幂等消费” 避免重复转发：

// 新增幂等处理逻辑：基于RocketMQ唯一消息ID去重
@Component
public class IdempotentConsumerHelper implements RocketMQPushConsumerListener {
    // ThreadLocal保存原始消息，用于获取唯一MsgId
    private static final ThreadLocal<MessageExt> MSG_HOLDER = new ThreadLocal<>();
    @Autowired
    private StringRedisTemplate redisTemplate;

    @Override
    public void prepareBeforeListen(MessageExt msgExt) {
        MSG_HOLDER.set(msgExt);
    }

    // 幂等校验：返回true表示未重复，false表示已重复
    public boolean checkIdempotent() {
        MessageExt msgExt = MSG_HOLDER.get();
        String msgId = msgExt.getMsgId(); // RocketMQ全局唯一消息ID
        // Redis记录已处理的消息ID，24小时过期
        Boolean isFirstConsume = redisTemplate.opsForValue().setIfAbsent(
                "mq:duplicate:" + msgId, "1", 24, TimeUnit.HOURS);
        return isFirstConsume != null && isFirstConsume;
    }
}

// 在消费者中集成幂等校验
@Component
public class RocketSpringConsumer implements RocketMQListener<String> {
    @Autowired
    private IdempotentConsumerHelper idempotentHelper;
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Override
    public void onMessage(String content) {
        // 第一步：幂等校验，避免重复消费
        if (!idempotentHelper.checkIdempotent()) {
            System.out.println("消息已处理，跳过重复消费：" + IdempotentConsumerHelper.getMsgId());
            return;
        }
        // 第二步：执行转发逻辑
        kafkaTemplate.send("kafka_data_topic", content).addCallback(...);
    }
}

• 额外建议：灰度期间新旧服务使用不同的 consumerGroup（如旧服务用rocket2kafka_old_group，新服务用rocket2kafka_new_group），避免 RocketMQ 负载均衡导致的重复拉取。

风险 3：Consumer Group 配置唯一性校验（解决方案：统一配置 + 启动校验）
Spring RocketMQ 要求同一个 consumerGroup 的所有实例必须使用完全一致的配置（如 topic、重试次数、消费模式等），否则会触发 “配置不一致” 异常。

• 配置统一管理：将 consumerGroup 的核心配置抽离到 Nacos 等配置中心，所有实例加载同一套配置。
• 启动前校验：应用启动时校验本地配置与 Broker 中已注册的 consumerGroup 配置是否一致，提前发现冲突。

3.2 消费组管理：原生编程式 vs Spring 注解式

消费组是 RocketMQ 的核心概念，支持两种创建 / 配置方式：

3.2.1 原生 Client：编程式创建 + 定制属性

通过MQAdminExt主动创建消费组并配置（适合运维 / 定制场景）：

public class RocketGroupAdminDemo {
    public static void main(String[] args) throws Exception {
        // 初始化RocketMQ Admin工具
        MQAdminExt mqAdmin = new MQAdminImpl(MixAll.DEFAULT_PRODUCER_GROUP, false, null);
        mqAdmin.setNamesrvAddr("localhost:9876");
        mqAdmin.start();

        try {
            // 配置消费组属性（如重试次数）
            ConsumerGroupConfig groupConfig = new ConsumerGroupConfig();
            groupConfig.setConsumeRetryTimes(3); // 最多重试3次
            // 主动创建消费组
            mqAdmin.createConsumerGroup(
                "programmatic_consumer_group",
                new ConsumerGroupInfo("programmatic_consumer_group", groupConfig)
            );
            System.out.println("原生Client创建消费组成功");
        } finally {
            mqAdmin.shutdown();
        }
    }
}

3.2.2 Spring Boot：注解式自动创建

只需在@RocketMQMessageListener中指定consumerGroup，Spring 启动时自动注册：

// Spring启动后自动创建消费组"spring_auto_consumer_group"
@Component
@RocketMQMessageListener(
    topic = "spring_biz_topic",
    consumerGroup = "spring_auto_consumer_group"
)
public class SpringAutoGroupConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String msg) {
        System.out.println("Spring自动消费组接收消息：" + msg);
    }
}

3.3 通用降级策略：双模式适配实现

降级是保障稳定性的关键，以下是两种方式的落地示例：

3.3.1 原生 Client：配置重试 + 非核心消息跳过

// 原生消费者配置降级
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("degrade_consumer_group");
consumer.setNamesrvAddr("localhost:9876");
consumer.subscribe("degrade_topic", "*");
// 1. 限制重试次数（超过则移入死信队列）
consumer.setMaxReconsumeTimes(3);
// 2. 消费时跳过非核心消息
consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
    for (MessageExt msg : msgs) {
        String content = new String(msg.getBody());
        if (isNonCoreMessage(content)) {
            System.out.println("原生降级：跳过非核心消息");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        // 核心消息业务处理
    }
    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
});
consumer.start();

3.3.2 Spring Boot：注解 + 配置中心开关降级

// Spring注解配置降级（重试次数+开关）
@Component
@RocketMQMessageListener(
    topic = "degrade_topic",
    consumerGroup = "spring_degrade_group",
    maxReconsumeTimes = 3 // 重试3次后移入死信
)
public class SpringDegradeConsumer implements RocketMQListener<String> {
    // 配置中心开关（Nacos/Apollo动态调整）
    @Value("${degrade.switch.non-core:false}")
    private boolean nonCoreSwitch;

    @Override
    public void onMessage(String content) {
        // 开关开启时，跳过非核心消息
        if (nonCoreSwitch && isNonCoreMessage(content)) {
            System.out.println("Spring降级：跳过非核心消息");
            return;
        }
        // 核心消息业务处理
    }
}

四、选型与总结

1、 选 Kafka：日志采集、大数据流式处理、高吞吐场景（百万级 / 秒）；
2、选 RocketMQ：电商 / 金融核心业务、跨系统消息同步、需要简洁 Spring 集成的场景；
3、RocketMQ 实现方式选择：

• 非 Spring 项目 / 需深度定制：用原生 Client；
• Spring Boot 项目 / 快速开发：用 Spring 集成版；

4、跨 MQ 同步：优先用 “消费 + 转发” 模式，注意手动提交 Offset 保证可靠性。