Spring集成消息队列实战教程-源码库

Spring集成消息队列实战教程：从配置到生产消费全流程

作为一名在微服务架构中摸爬滚打多年的开发者，我深知消息队列在系统解耦、流量削峰中的重要性。今天就来分享我在Spring项目中集成RabbitMQ的完整实战经验，包含那些官方文档不会告诉你的踩坑细节。

环境准备与依赖配置

首先确保你的项目使用Spring Boot 2.7+版本，在pom.xml中添加以下依赖：


  org.springframework.boot
  spring-boot-starter-amqp

在application.yml中配置RabbitMQ连接信息时，我强烈建议设置connection-timeout，这在网络不稳定的生产环境中能避免很多诡异问题：

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: guest
    password: guest
    connection-timeout: 5s
    template:
      retry:
        enabled: true
        max-attempts: 3

队列与交换机配置

直接通过注解声明队列和交换机是最优雅的方式。这里我创建了一个直连交换机和一个持久化队列：

@Configuration
public class RabbitMQConfig {
    
    @Bean
    public Queue orderQueue() {
        return new Queue("order.queue", true); // true表示持久化
    }
    
    @Bean
    public DirectExchange orderExchange() {
        return new DirectExchange("order.exchange");
    }
    
    @Bean
    public Binding binding(Queue orderQueue, DirectExchange orderExchange) {
        return BindingBuilder.bind(orderQueue)
               .to(orderExchange)
               .with("order.routingKey");
    }
}

消息生产者实现

使用RabbitTemplate发送消息时，记得要设置消息的持久化属性。这是我曾经踩过的坑——默认情况下消息是非持久化的：

@Service
public class OrderMessageProducer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    public void sendOrderMessage(OrderDTO order) {
        MessageProperties properties = new MessageProperties();
        properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
        
        Message message = new Message(
            JSON.toJSONBytes(order), 
            properties
        );
        
        rabbitTemplate.convertAndSend(
            "order.exchange",
            "order.routingKey", 
            message
        );
        
        log.info("订单消息发送成功，订单号：{}", order.getOrderNo());
    }
}

消息消费者实现

使用@RabbitListener注解消费消息时，务必处理异常情况。我推荐手动确认模式，只有在业务处理成功后才确认消息：

@Component
public class OrderMessageConsumer {
    
    @RabbitListener(queues = "order.queue")
    public void processOrder(Message message, Channel channel) throws IOException {
        try {
            OrderDTO order = JSON.parseObject(
                message.getBody(), 
                OrderDTO.class
            );
            
            // 业务处理逻辑
            orderService.processOrder(order);
            
            // 手动确认消息
            channel.basicAck(
                message.getMessageProperties().getDeliveryTag(), 
                false
            );
            
        } catch (Exception e) {
            log.error("订单处理失败", e);
            // 拒绝消息并重新入队
            channel.basicNack(
                message.getMessageProperties().getDeliveryTag(),
                false, 
                true
            );
        }
    }
}