Spring Boot中RabbitMQ的六种工作方式及应用场景与代码示例

Spring Boot中RabbitMQ的六种工作方式及应用场景与代码示例

在现代分布式系统中，消息队列扮演着至关重要的角色。RabbitMQ作为一款功能强大的开源消息中间件，广泛应用于异步处理、解耦系统、流量削峰等场景。本文将详细介绍在Spring Boot中使用RabbitMQ的六种工作模式，并结合实际代码示例和应用场景进行讲解。

一、RabbitMQ六种工作模式简介

RabbitMQ支持多种消息传递模式，每种模式适用于不同的业务需求。以下是六种主要的工作模式：

二、工作模式详解与代码示例

1. Simple（简单模式）

特点：一个生产者对应一个消费者，消息直接发送到队列。

应用场景：简单的任务分发，如发送邮件通知。

配置文件（application.yml）

spring:
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest

生产者代码示例（Producer.java）

@Component
public class SimpleProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

public void send(String message) {
rabbitTemplate.convertAndSend("simple_queue", message);
System.out.println("[Simple] 消息已发送: " + message);
}
}

消费者代码示例（Consumer.java）

@Component
@RabbitListener(queues = "simple_queue")
public class SimpleConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Simple] 收到消息: " + message);
}
}

2. Work Queue（工作队列模式）

特点：多个消费者竞争消费同一个队列中的消息，实现负载均衡。

应用场景：耗时任务处理，如图片压缩、日志处理。

消费者代码示例（WorkerConsumer.java）

@Component
@RabbitListener(queues = "work_queue")
public class WorkerConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Work Queue] 消费者处理消息: " + message + " (线程: " + Thread.currentThread().getName() + ")");
try {
Thread.sleep(2000); // 模拟耗时操作
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}

注意：需要启动多个消费者实例以验证负载均衡效果。

3. Publish/Subscribe（发布/订阅模式）

特点：一个消息被多个消费者接收，通过交换机（fanout）广播给所有绑定的队列。

应用场景：实时通知、广播消息。

配置类（RabbitConfig.java）

@Configuration
public class RabbitConfig {

@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange("fanout_exchange");
}

@Bean
public Queue queueA() {
return new Queue("queue_a");
}

@Bean
public Queue queueB() {
return new Queue("queue_b");
}

@Bean
public Binding bindingA() {
return BindingBuilder.bind(queueA()).to(fanoutExchange());
}

@Bean
public Binding bindingB() {
return BindingBuilder.bind(queueB()).to(fanoutExchange());
}
}

生产者代码示例（Publisher.java）

@Component
public class Publisher {

@Autowired
private RabbitTemplate rabbitTemplate;

public void sendMessage(String message) {
rabbitTemplate.convertAndSend("fanout_exchange", "", message);
System.out.println("[Publish/Subscribe] 广播消息: " + message);
}
}

消费者代码示例（Subscriber.java）

@Component
@RabbitListener(queues = "queue_a")
public class SubscriberA {

@RabbitHandler
public void receive(String message) {
System.out.println("[Subscriber A] 接收消息: " + message);
}
}

@Component
@RabbitListener(queues = "queue_b")
public class SubscriberB {

@RabbitHandler
public void receive(String message) {
System.out.println("[Subscriber B] 接收消息: " + message);
}
}

4. Routing（路由模式）

特点：基于路由键（routing key）将消息发送到特定队列，交换机为direct。

应用场景：根据类型分发消息，如不同类型的日志处理。

配置类（RabbitConfig.java）

@Bean
public DirectExchange directExchange() {
return new DirectExchange("direct_exchange");
}

@Bean
public Queue errorQueue() {
return new Queue("error_queue");
}

@Bean
public Queue infoQueue() {
return new Queue("info_queue");
}

@Bean
public Binding bindingError() {
return BindingBuilder.bind(errorQueue()).to(directExchange()).with("error");
}

@Bean
public Binding bindingInfo() {
return BindingBuilder.bind(infoQueue()).to(directExchange()).with("info");
}

生产者代码示例（RoutingProducer.java）

@Component
public class RoutingProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

public void send(String routingKey, String message) {
rabbitTemplate.convertAndSend("direct_exchange", routingKey, message);
System.out.println("[Routing] 发送消息到路由键: " + routingKey + " – " + message);
}
}

消费者代码示例（RoutingConsumer.java）

@Component
@RabbitListener(queues = "error_queue")
public class ErrorConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Error Consumer] 处理错误消息: " + message);
}
}

@Component
@RabbitListener(queues = "info_queue")
public class InfoConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Info Consumer] 处理信息消息: " + message);
}
}

5. Topics（主题模式）

特点：支持通配符匹配，灵活地将消息分发到多个队列。

应用场景：复杂的事件分发，如订单状态变更通知。

配置类（RabbitConfig.java）

@Bean
public TopicExchange topicExchange() {
return new TopicExchange("topic_exchange");
}

@Bean
public Queue orderQueue() {
return new Queue("order_queue");
}

@Bean
public Queue paymentQueue() {
return new Queue("payment_queue");
}

@Bean
public Binding bindingOrder() {
return BindingBuilder.bind(orderQueue()).to(topicExchange()).with("order.#");
}

@Bean
public Binding bindingPayment() {
return BindingBuilder.bind(paymentQueue()).to(topicExchange()).with("payment.*");
}

生产者代码示例（TopicProducer.java）

@Component
public class TopicProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

public void send(String routingKey, String message) {
rabbitTemplate.convertAndSend("topic_exchange", routingKey, message);
System.out.println("[Topics] 发送消息到路由键: " + routingKey + " – " + message);
}
}

消费者代码示例（TopicConsumer.java）

@Component
@RabbitListener(queues = "order_queue")
public class OrderConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Order Consumer] 接收订单消息: " + message);
}
}

@Component
@RabbitListener(queues = "payment_queue")
public class PaymentConsumer {

@RabbitHandler
public void receive(String message) {
System.out.println("[Payment Consumer] 接收支付消息: " + message);
}
}

6. Headers（头信息模式）

特点：通过消息头（headers）匹配规则，而不是路由键，适合复杂条件匹配。

应用场景：基于多维度属性的消息过滤，如按用户类型、区域等筛选。

配置类（RabbitConfig.java）

@Bean
public HeadersExchange headersExchange() {
return new HeadersExchange("headers_exchange");
}

@Bean
public Queue userQueue() {
return new Queue("user_queue");
}

@Bean
public Binding bindingUser() {
Map<String, Object> map = new HashMap<>();
map.put("user_type", "vip");
map.put("region", "china");
return BindingBuilder.bind(userQueue()).to(headersExchange()).whereAll(map).match();
}

生产者代码示例（HeadersProducer.java）

@Component
public class HeadersProducer {

@Autowired
private RabbitTemplate rabbitTemplate;

public void send(String message) {
MessageProperties properties = new MessageProperties();
properties.setHeader("user_type", "vip");
properties.setHeader("region", "china");
Message msg = new Message(message.getBytes(), properties);
rabbitTemplate.send("headers_exchange", null, msg);
System.out.println("[Headers] 发送带头信息的消息: " + message);
}
}

消费者代码示例（HeadersConsumer.java）

@Component
@RabbitListener(queues = "user_queue")
public class HeadersConsumer {

@RabbitHandler
public void receive(Message message) {
String msg = new String(message.getBody());
System.out.println("[Headers Consumer] 接收消息: " + msg);
}
}

三、总结

通过合理选择RabbitMQ的工作模式，可以显著提升系统的可扩展性与灵活性。建议根据实际业务需求选择最合适的模式，并结合Spring Boot的自动化配置简化开发流程。

提示：在生产环境中，请务必配置合理的消息确认机制（ACK）、持久化以及死信队列，以确保消息可靠性。