在现代微服务架构中，消息队列（Message Queue，MQ）扮演着至关重要的角色，它能够帮助我们实现异步处理、流量削峰、系统解耦等功能。本文将以 Spring Boot 为例，深入探讨如何利用 MQ 实现这些强大的特性，并结合实际场景，给出详细的代码实现和步骤说明。
一、项目准备
首先，我们需要创建一个 Spring Boot 项目，并引入相关依赖：
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency>
</dependencies>
这里我们引入了 spring-boot-starter-amqp 来支持 RabbitMQ。
二、异步处理
传统的同步调用方式，调用方需要等待被调用方执行完成后才能继续执行。而使用 MQ 进行异步处理，调用方只需将消息发送到队列，然后继续执行其他任务，从而提高系统响应速度。
2.1 定义消息模型
@Data
public class OrderMessage {private Long orderId;private String productName;private BigDecimal price;
}
2.2 配置消息队列和交换机
@Configuration
public class RabbitMqConfig {public static final String ORDER_QUEUE = "order.queue";public static final String ORDER_EXCHANGE = "order.exchange";@Beanpublic Queue orderQueue() {return new Queue(ORDER_QUEUE, true);}@Beanpublic DirectExchange orderExchange() {return new DirectExchange(ORDER_EXCHANGE);}@Beanpublic Binding bindingOrder() {return BindingBuilder.bind(orderQueue()).to(orderExchange()).with("order.routing.key");}
}
这里我们定义了一个名为 order.queue 的队列，一个名为 order.exchange 的 Direct 交换机，并将它们绑定在一起。
2.3 发送消息
@Service
public class OrderService {@Autowiredprivate RabbitTemplate rabbitTemplate;public void createOrder(OrderMessage message) {// ... 处理订单逻辑 ...rabbitTemplate.convertAndSend(RabbitMqConfig.ORDER_EXCHANGE, "order.routing.key", message);}
}
在 OrderService 中，我们使用 RabbitTemplate 将订单消息发送到 order.exchange 交换机。
2.4 消费消息
@Component
@RabbitListener(queues = RabbitMqConfig.ORDER_QUEUE)
public class OrderConsumer {@RabbitHandlerpublic void handleOrderMessage(OrderMessage message) {// ... 处理订单消息 ...System.out.println("Received order message: " + message);}
}
使用 @RabbitListener 注解监听 order.queue 队列，当有消息到达时，handleOrderMessage() 方法会被自动调用。
三、流量削峰
当系统面临突发流量冲击时，MQ 可以充当缓冲区，将请求先存储在队列中，然后逐步消费，避免系统过载。
3.1 配置消息监听器
@Component
@RabbitListener(queues = RabbitMqConfig.ORDER_QUEUE, concurrency = "5-10")
public class OrderConsumer {// ...
}
通过设置 concurrency 属性，我们可以控制消息监听器的并发数量，从而控制消息的消费速度，达到流量削峰的目的。
四、消息总线
消息总线（Message Bus）是一种设计模式，它允许不同的组件通过共享的消息通道进行通信。
4.1 创建主题交换机
@Configuration
public class RabbitMqConfig {public static final String EVENT_EXCHANGE = "event.exchange";@Beanpublic TopicExchange eventExchange() {return new TopicExchange(EVENT_EXCHANGE);}
}
这里我们创建了一个名为 event.exchange 的主题交换机，用于实现消息总线。
4.2 发布消息
@Service
public class EventPublisher {@Autowiredprivate RabbitTemplate rabbitTemplate;public void publishEvent(String event, Object data) {rabbitTemplate.convertAndSend(RabbitMqConfig.EVENT_EXCHANGE, event, data);}
}
4.3 订阅消息
@Component
@RabbitListener(bindings = @RabbitListenerBinding(value = @QueueBinding(value = @Queue(value = "user.queue", durable = "true"),exchange = @Exchange(value = RabbitMqConfig.EVENT_EXCHANGE, type = ExchangeTypes.TOPIC),key = "user.#")
))
public class UserEventListener {@RabbitHandlerpublic void handleUserEvent(UserEvent event) {// ... 处理用户事件 ...}
}
使用 user.# 作为路由键，UserEventListener 将会接收到所有以 user. 开头的事件消息。
五、延迟队列
延迟队列允许我们设置消息的延迟时间，消息在延迟时间到达后才会被消费者消费。
5.1 创建延迟队列
@Configuration
public class RabbitMqConfig {public static final String DELAY_QUEUE = "delay.queue";public static final String DELAY_EXCHANGE = "delay.exchange";@Beanpublic Queue delayQueue() {return QueueBuilder.durable(DELAY_QUEUE).withArgument("x-dead-letter-exchange", "dead.letter.exchange") // 设置死信交换机.withArgument("x-dead-letter-routing-key", "dead.letter.routing.key") // 设置死信路由键.build();}@Beanpublic DirectExchange delayExchange() {return new DirectExchange(DELAY_EXCHANGE);}@Beanpublic Binding bindingDelay() {return BindingBuilder.bind(delayQueue()).to(delayExchange()).with("delay.routing.key");}
}
5.2 发送延迟消息
@Service
public class DelayMessageSender {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendDelayMessage(Object message, long delayTime) {rabbitTemplate.convertAndSend(RabbitMqConfig.DELAY_EXCHANGE, "delay.routing.key", message, msg -> {msg.getMessageProperties().setExpiration(String.valueOf(delayTime));return msg;});}
}
5.3 处理延迟消息
我们需要创建一个新的队列和交换机来处理死信消息：
@Configuration
public class RabbitMqConfig {// ...@Beanpublic Queue deadLetterQueue() {return new Queue("dead.letter.queue", true);}@Beanpublic DirectExchange deadLetterExchange() {return new DirectExchange("dead.letter.exchange");}@Beanpublic Binding bindingDeadLetter() {return BindingBuilder.bind(deadLetterQueue()).to(deadLetterExchange()).with("dead.letter.routing.key");}
}@Component
@RabbitListener(queues = "dead.letter.queue")
public class DeadLetterConsumer {@RabbitHandlerpublic void handleDeadLetterMessage(String message) {// ... 处理延迟消息 ...System.out.println("Received dead letter message: " + message);}
}
六、广播消息推送
广播消息允许我们将同一条消息发送给多个消费者。
6.1 创建扇形交换机
@Configuration
public class RabbitMqConfig {public static final String FANOUT_EXCHANGE = "fanout.exchange";@Beanpublic FanoutExchange fanoutExchange() {return new FanoutExchange(FANOUT_EXCHANGE);}
}
6.2 发送广播消息
@Service
public class BroadcastMessageSender {@Autowiredprivate RabbitTemplate rabbitTemplate;public void sendBroadcastMessage(Object message) {rabbitTemplate.convertAndSend(RabbitMqConfig.FANOUT_EXCHANGE, "", message);}
}
6.3 接收广播消息
每个消费者都需要创建一个独立的队列，并绑定到 fanout.exchange 交换机：
@Component
@RabbitListener(bindings = @RabbitListenerBinding(value = @QueueBinding(value = @Queue(value = "broadcast.queue.1", durable = "true"),exchange = @Exchange(value = RabbitMqConfig.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT))
))
public class BroadcastConsumer1 {@RabbitHandlerpublic void handleBroadcastMessage(String message) {// ... 处理广播消息 ...System.out.println("Consumer 1 received broadcast message: " + message);}
}@Component
@RabbitListener(bindings = @RabbitListenerBinding(value = @QueueBinding(value = @Queue(value = "broadcast.queue.2", durable = "true"),exchange = @Exchange(value = RabbitMqConfig.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT))
))
public class BroadcastConsumer2 {@RabbitHandlerpublic void handleBroadcastMessage(String message) {// ... 处理广播消息 ...System.out.println("Consumer 2 received broadcast message: " + message);}
}
七、总结
本文介绍了如何使用 Spring Boot 和 RabbitMQ 实现异步处理、流量削峰、消息总线、延迟队列和广播消息推送等功能。希望这篇文章能够帮助你更好地理解 MQ 的强大功能，并在实际项目中灵活运用。