Spring Boot 整合 RabbitMQ从入门到实战步骤

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:management

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

spring:
  rabbitmq:
    host: localhost       # RabbitMQ 服务器地址
    port: 5672           # 默认端口
    username: guest      # 默认用户名
    password: guest      # 默认密码
    virtual-host: /      # 虚拟主机（默认）
    publisher-confirm-type: correlated  # 开启消息确认
    publisher-returns: true             # 开启消息失败回调
    listener:
      simple:
        acknowledge-mode: manual       # 手动确认消息
        prefetch: 1                    # 每次预取的消息数

@Configuration
public class RabbitConfig {
    // 直连交换机
    public static final String DIRECT_EXCHANGE = "direct_exchange";
    // 队列
    public static final String DIRECT_QUEUE = "direct_queue";
    // 路由键
    public static final String ROUTING_KEY = "routing_key";
    @Bean
    public Exchange directExchange() {
        return ExchangeBuilder.directExchange(DIRECT_EXCHANGE).build();
    }
    @Bean
    public Queue directQueue() {
        return new Queue(DIRECT_QUEUE, true); // durable=true 表示持久化
    }
    @Bean
    public Binding binding(Queue queue, Exchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);
    }
}

@Service
public class RabbitProducer {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    public void sendMessage(String message) {
        rabbitTemplate.convertAndSend(
            RabbitConfig.DIRECT_EXCHANGE, 
            RabbitConfig.ROUTING_KEY, 
            message
        );
    }
}

@Service
public class RabbitConsumer {
    @RabbitListener(queues = RabbitConfig.DIRECT_QUEUE)
    public void receiveMessage(String message) {
        System.out.println("收到消息: " + message);
        // 手动确认消息（需配置 acknowledge-mode: manual）
        // 实际项目中需处理异常和重试逻辑
    }
}

@Bean
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    RabbitTemplate template = new RabbitTemplate(connectionFactory);
    template.setConfirmCallback((correlationData, ack, cause) -> {
        if (ack) {
            System.out.println("消息确认成功");
        } else {
            System.err.println("消息确认失败: " + cause);
        }
    });
    template.setMandatory(true); // 开启强制回调
    return template;
}

@RabbitListener(queues = RabbitConfig.DIRECT_QUEUE)
public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
    try {
        System.out.println("处理消息: " + message);
        channel.basicAck(deliveryTag, false); // 确认消息
    } catch (Exception e) {
        channel.basicNack(deliveryTag, false, true); // 拒绝消息并重新入队
    }
}

@Bean
public Queue deadLetterQueue() {
    Map<String, Object> args = new HashMap<>();
    args.put("x-dead-letter-exchange", "dlx_exchange"); // 指定死信交换机
    args.put("x-dead-letter-routing-key", "dlx_routing_key");
    return new Queue("dlx_queue", true, false, false, args);
}

@SpringBootTest
@AutoConfigureMockMvc
public class RabbitMQApplicationTests {
    @Autowired
    private RabbitProducer producer;
    @Test
    public void testSendMessage() {
        producer.sendMessage("Hello RabbitMQ!");
    }
}