Spring Boot 3.5.11 + JDK21 整合 RabbitMQ、RocketMQ、Kafka 实战与选型

application.yml

spring:
  threads:
    virtual:
      enabled: true # 开启虚拟线程支持
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest

配置虚拟线程执行器 这里我们自定义一个任务执行器来利用 JDK21 的虚拟线程特性，避免阻塞 IO 操作占用平台线程。

@Bean
public Executor rabbitVirtualExecutor() {
    return new org.springframework.core.task.VirtualThreadTaskExecutor("rabbit-virtual-");
}

@Bean
public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(
        ConnectionFactory connectionFactory, Executor rabbitVirtualExecutor) {
    SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
    factory.setConnectionFactory(connectionFactory);
    factory.setTaskExecutor(rabbitVirtualExecutor);
    factory.setConcurrentConsumers(5);
    factory.setMaxConcurrentConsumers(20);
    return factory;
}

生产者 + 消费者

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

    @Bean
    CommandLineRunner runner(RabbitTemplate template) {
        return args -> {
            template.convertAndSend("test-queue", "Hello RabbitMQ");
        };
    }

    @RabbitListener(queuesToDeclare = @Queue("test-queue"), containerFactory = "rabbitListenerContainerFactory")
    public void consume(String msg) {
        System.out.println("收到：" + msg);
        System.out.println("虚拟线程：" + Thread.currentThread().isVirtual());
    }
}

2. 整合 RocketMQ

依赖

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>2.3.2</version>
</dependency>

application.yml

spring:
  threads:
    virtual:
      enabled: true
  rocketmq:
    name-server: localhost:9876
    producer:
      group: test-producer-group

生产者

@Autowired
private RocketMQTemplate rocketMQTemplate;

@Bean
CommandLineRunner rocketRunner() {
    return args -> {
        rocketMQTemplate.convertAndSend("test-topic", "Hello RocketMQ");
    };
}

消费者

@RocketMQMessageListener(
    topic = "test-topic",
    consumerGroup = "test-consumer-group"
)
@Component
public class RocketConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.println("收到：" + message);
        System.out.println("虚拟线程：" + Thread.currentThread().isVirtual());
    }
}

RocketMQ 的线程模型由客户端内部管理，如需定制线程池，应通过对应配置项或扩展方式调整。

3. 整合 Kafka

依赖

<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

application.yml

spring:
  threads:
    virtual:
      enabled: true
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: test-group
      auto-offset-reset: earliest

Kafka Listener 容器工厂

@Bean
public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
        ConsumerFactory<String, String> cf) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(cf);
    factory.setConcurrency(5);
    return factory;
}

生产者 + 消费者

@Autowired
private KafkaTemplate<String, String> kafkaTemplate;

@Bean
CommandLineRunner kafkaRunner() {
    return args -> kafkaTemplate.send("test-topic", "Hello Kafka");
}

@KafkaListener(topics = "test-topic", containerFactory = "kafkaListenerContainerFactory")
public void listen(String message) {
    System.out.println("收到：" + message);
    System.out.println("虚拟线程：" + Thread.currentThread().isVirtual());
}

五、生产级优化建议

RabbitMQ

• 开启生产者确认机制
• 合理设置 prefetch
• 使用持久化队列 + 持久化消息
• 监控积压与连接数

RocketMQ

• 合理设置刷盘策略（SYNC_FLUSH / ASYNC_FLUSH）
• 核心业务使用事务消息
• Broker 主从部署

Kafka

• 分区数量与副本因子合理规划
• 使用事务保证 Exactly-Once
• 启用批量消费提升吞吐

六、最终选型总结

七、结语

在 Spring Boot 3.5.11 + JDK21 技术栈下：

• 虚拟线程降低并发成本
• 三大 MQ 均可稳定集成
• 选型关键在业务模型，而非'谁吞吐更高'

架构没有绝对优劣，只有是否适合你的系统。