Spring Boot 与 RabbitMQ：从 Hello World 到生产环境实战

然后在 application.yml 里把连接信息填上：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    virtual-host: /

Virtual-host 可以理解成命名空间，默认 / 就够用。Spring Boot 会根据这些配置自动创建 ConnectionFactory、RabbitTemplate、RabbitAdmin 这些核心 bean。

发消息：用 RabbitTemplate

写个 Controller，直接注入 RabbitTemplate：

// HelloController.java
package com.example.demo;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class HelloController {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/send")
    public String sendMessage(@RequestParam String msg) {
        rabbitTemplate.convertAndSend("hello.queue", msg);
        return "Message sent: " + msg;
    }
}

调用 convertAndSend 时，第一个参数是 routing key，因为用了默认 Direct Exchange，它和队列名一样。第二个参数可以是任意对象，Spring 会帮你序列化成字节数组。

收消息：@RabbitListener

消费者的写法更简单：

// HelloListener.java
package com.example.demo;

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class HelloListener {

    @RabbitListener(queues = "hello.queue")
    public void receiveMessage(String message) {
        System.out.println("【收到消息】: " + message);
    }
}

queues 属性指定要监听哪个队列。方法参数能自动反序列化成 String、byte[] 或 Serializable 对象，默认用的是 SimpleMessageConverter。

顺序说一下：启动应用，浏览器访问 /send?msg=Hello%20RabbitMQ，控制台就会打印：'【收到消息】: Hello RabbitMQ'。同时，RabbitMQ 的管理界面里会凭空多出一个 hello.queue——因为 RabbitAdmin 会偷偷帮你声明队列。

这种自动声明虽然爽，但线上还是应该显式定义队列，把持久化这些属性定死，省得重启之后队列消失。

显式声明队列

加个 @Configuration：

// RabbitConfig.java
package com.example.demo;

import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    @Bean
    public Queue helloQueue() {
        return new Queue("hello.queue", true); // 持久化
    }
}

第二个参数设成 true 表示队列会持久化，RabbitMQ 重启后还在。其他的构造参数还可以控制排他性、自动删除之类，按需选用。

消息序列化：别用 Java 原生的

默认的 SimpleMessageConverter 对 String 和 byte[] 处理还行，但碰到自定义对象就会走 Java 序列化。这东西既不可读，版本兼容性也差。所以基本上项目一启动，我就会把转换器换成 JSON 的。

// RabbitConfig.java 里追加
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;

@Bean
public MessageConverter jsonMessageConverter() {
    return new Jackson2JsonMessageConverter();
}

这样就能直接发送对象了：

public class Greeting {
    private String content;
    public Greeting(String content) { this.content = content; }
    public String getContent() { return content; }
}

@GetMapping("/sendObj")
public String sendObject() {
    Greeting greeting = new Greeting("Hello from Object!");
    rabbitTemplate.convertAndSend("hello.queue", greeting);
    return "Object sent";
}

Listener 里接收时参数类型直接写 Greeting 就行，Spring 会自动做反序列化。

消息确认，得手动来

默认情况下消费者是自动确认的：消息一投递给处理方法，RabbitMQ 就当它处理完了，立刻把消息从队列里删掉。万一方法执行到一半崩了，消息就丢了。

生产环境一定得改成手动确认：

spring:
  rabbitmq:
    listener:
      simple:
        acknowledge-mode: manual

然后 Listener 里要显式 ack：

import com.rabbitmq.client.Channel;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;

@RabbitListener(queues = "hello.queue")
public void receiveMessage(String message, Channel channel,
                           @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    try {
        System.out.println("处理消息：" + message);
        if ("error".equals(message)) {
            throw new RuntimeException("模拟异常");
        }
        channel.basicAck(deliveryTag, false);
    } catch (Exception e) {
        try {
            channel.basicNack(deliveryTag, false, false); // 拒绝，不重新入队
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }
}

basicAck 告诉 RabbitMQ 消息处理完了，可以删。basicNack 的第三个参数 requeue 设成 false 表示不重新入队，不然会死循环。

重试与死信

手动确认解决了丢消息的问题，但临时的故障（比如数据库连不上）直接 nack 又太粗暴。利用 Spring 自带的重试可以扛一会儿：

spring:
  rabbitmq:
    listener:
      simple:
        retry:
          enabled: true
          max-attempts: 3
          initial-interval: 1000

重试 3 次，间隔 1 秒。要是还不行，就扔进死信队列等着人工处理。

配置死信需要声明对应的 Exchange 和队列，并让原队列绑定死信参数：

@Bean
public Queue dlq() {
    return QueueBuilder.durable("hello.dlq").build();
}

@Bean
public DirectExchange dlx() {
    return new DirectExchange("hello.dlx");
}

@Bean
public Binding dlqBinding() {
    return BindingBuilder.bind(dlq()).to(dlx()).with("hello.dlq");
}

@Bean
public Queue helloQueue() {
    return QueueBuilder.durable("hello.queue")
            .withArgument("x-dead-letter-exchange", "hello.dlx")
            .withArgument("x-dead-letter-routing-key", "hello.dlq")
            .build();
}

现在可以另写一个监听器专盯死信队列，记日志、发报警都随你。

并发消费者

默认情况下，一个 @RabbitListener 就一条线程在干活。消息积压的时候可以调高并发：

spring:
  rabbitmq:
    listener:
      simple:
        concurrency: 5
        max-concurrency: 10

concurrency 是初始线程数，max-concurrency 是上限。Spring 会根据负载动态伸缩，但伸缩本身也有开销，别设得太离谱。

单元测试

用 Testcontainers 拉起一个真实的 RabbitMQ 容器做集成测试，比 mock 靠谱得多。

@SpringBootTest
@Testcontainers
class RabbitMQIntegrationTest {

    @Container
    static RabbitMQContainer rabbitMQ = new RabbitMQContainer("rabbitmq:3-management");

    @DynamicPropertySource
    static void configureProperties(DynamicPropertyRegistry registry) {
        registry.add("spring.rabbitmq.host", rabbitMQ::getHost);
        registry.add("spring.rabbitmq.port", rabbitMQ::getAmqpPort);
        registry.add("spring.rabbitmq.username", rabbitMQ::getAdminUsername);
        registry.add("spring.rabbitmq.password", rabbitMQ::getAdminPassword);
    }

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    void testSendMessage() throws InterruptedException {
        rabbitTemplate.convertAndSend("test.queue", "Hello Test");
        Thread.sleep(1000); // 等等消费者
        // 这里验证结果，比如查数据库
    }
}

常见问题

• 消息发到了但消费者没反应：队列名对得上吗？去管理后台看看消息是不是还在队列里。可能 Listener 没被 Spring 扫描到，记得加 @Component。
• 启动报连接拒绝：RabbitMQ 服务到底起来了没？防火墙有没有拦住 5672？用户名密码和权限也要检查。
• 乱码或反序列化失败：生产者和消费者用的 MessageConverter 必须一致。JSON 的话，确保两边都有 Jackson，对象字段名匹配。
• 消息丢失：队列、消息都要持久化，手动确认后必须 ack，别光吃消息不认账。

和其他 MQ 比一下

RabbitMQ 的路由灵活，延迟低，适合订单、通知这类业务。要追求超高吞吐和顺序性，Kafka 可能更合适。

不止点对点

如果要做广播，换成 FanoutExchange，把几个队列绑上去就好：

@Bean
public FanoutExchange fanoutExchange() {
    return new FanoutExchange("broadcast.exchange");
}

@Bean
public Queue emailQueue() {
    return new Queue("email.queue");
}

@Bean
public Queue smsQueue() {
    return new Queue("sms.queue");
}

@Bean
public Binding bindEmail() {
    return BindingBuilder.bind(emailQueue()).to(fanoutExchange());
}

@Bean
public Binding bindSms() {
    return BindingBuilder.bind(smsQueue()).to(fanoutExchange());
}

发送时只需指定 exchange 名字：

rabbitTemplate.convertAndSend("broadcast.exchange", "", "Hello All!");

最后

这个 Hello World 示范了最简流程，然后一路加到了序列化、确认、重试、死信这些生产上跑不掉的配置。消息队列确实好用，但一定要把持久化、确认这些保底的措施搞清楚，别等丢消息了再回头补。

后续还可以接上链路追踪、Prometheus 监控、集群高可用，一步步让系统更扛造。