RabbitMQ 七种工作模式

前言

RabbitMQ 是目前最流行的开源消息代理之一，基于 AMQP（高级消息队列协议）实现，支持多种消息传递模式。它凭借高可靠、易扩展、多语言客户端等特性，成为微服务、分布式系统、异步任务处理的核心组件。本篇通过Java代码示例，讲解RabbitMQ七种工作模式，它的工作原理以及在springboot项目中如何使用

RabbitMQ 七种工作模式

1.简单模式

点对点，一个生产者，一个消费者

最简单的点对点通信模型。一个生产者将消息发送到队列，一个消费者从该队列中获取消息。底层使用 RabbitMQ 默认交换机（空字符串名称，类型为 direct），并自动绑定到队列（routing key 等于队列名）

• 邮件通知、短信发送等简单的异步任务。

• 开发环境快速验证消息收发逻辑。

生产者代码

public class Producer {
public static void main(String[] args) throws IOException, TimeoutException {
//1.建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost("virtualHost01");//虚拟主机
connectionFactory.setPort(5672);//端口号
connectionFactory.setHost("");//ip
connectionFactory.setPassword("123456");
connectionFactory.setUsername("winx");
Connection connection=connectionFactory.newConnection();//创建 连接
//2.创建 信道
Channel channel=connection.createChannel();
//3.声明 交换机(在此不创建 使用内置的交换机)
//4.声明队列
channel.queueDeclare("hello",true,false,false,null);
/**
* queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,
* Map<String, Object> arguments)
* queue :队列名称
* durable:是否可持久化
* exclusive:是否独占
* autoDelete:是否自动删除
* arguments:参数
*
*/
//5.传递消息
String msg="hello,RabbitMQ";
channel.basicPublish("","hello",null,msg.getBytes());
/**
* basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)
* exchange:交换机名称
* routingKey:交换机与队列建立联系的，内置交换机🀄 routingKey与队列名称相同
* props：属性配置
* body:要发送的消息
*/
//6.释放资源
System.out.println("消息发送成功!");
channel.close();
connection.close();
}
}

消费者代码

public class Customer {
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//1.连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("");
connectionFactory.setUsername("winx");
connectionFactory.setPassword("123456");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("virtualHost01");
//2.创建channel
Connection connection=connectionFactory.newConnection();
Channel channel=connection.createChannel();
//3.声明队列
channel.queueDeclare("hello",true,false,false,null);

//4.消费消息
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println("接收到消息"+new String(body));
}
};
channel.basicConsume("hello",true,consumer);
/**
* basicConsume(String queue, boolean autoAck, Consumer callback
* queue 队列名称
* autoAck：是否自动确认
* callback：对消息具体的执行逻辑
*
*/
//等待程序打印完毕
Thread.sleep(2000);

//5.释放资源
channel.close();
connection.close();

}
}

2.工作队列

一个生产者，多个消费者，消息不重复消费

一个队列被多个消费者监听，消息以轮询分发（Round-Robin）的方式均匀分发给空闲的消费者。该模式适用于任务量较大且可并行处理的场景，能够有效分摊负载。

• 图片/视频处理、报表生成等耗时任务的后台执行。

• 多个 worker 实例并发处理同一类任务

与简单模式相似，启动多个消费者就可以达成

public class Customer1 {
public static void main(String[] args) throws IOException, TimeoutException {
//1.建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost(Constant.VirtualHost);
connectionFactory.setHost(Constant.Host);
connectionFactory.setPort(Constant.Port);
connectionFactory.setUsername(Constant.Username);
connectionFactory.setPassword(Constant.Password);
Connection connection=connectionFactory.newConnection();
//2.创建channel
Channel channel=connection.createChannel();
//3.声明交换机
//使用默认无需声明
//4.声明队列
channel.queueDeclare(Constant.Work_Queue,true,false,false,null);
//5.消费信息
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
System.out.println(new String(body));
}
};
channel.basicConsume(Constant.Work_Queue,true,consumer);
//6.释放资源
channel.close();
connection.close();
}

}

在SpringBoot中声明多个监听队列的消费者

@Component
public class WorkListener {

@RabbitListener(queues = Constant.WORK_QUEUE)
public void Listener1(Message message){
System.out.println("listen1"+"接收到消息"+"["+Constant.WORK_QUEUE+"]"+message);
}

@RabbitListener(queues = Constant.WORK_QUEUE)
public void Listener2(Message message){
System.out.println("listen2"+"接收到消息"+"["+Constant.WORK_QUEUE+"]"+message);
}
}

3.发布订阅模式

rountingKey bountingKey

一个生产者，多个消费者，交换机消息复制多份，每个消费者接收到相同的消息

引入 Fanout Exchange（扇出交换机），生产者将消息发送到交换机，交换机将消息广播到所有绑定到此交换机的队列。每个队列对应一个消费者（或消费组），所有消费者都会收到相同的消息副本。

使用场景：

• 一个数据变更需要同时更新多个下游服务（如缓存刷新、搜索引擎索引更新）。

• 实时推送消息到多个客户端

生产者

channel.exchangeDeclare(Constant.Exchange, BuiltinExchangeType.FANOUT,true);
//4.声明队列
channel.queueDeclare(Constant.Pubsub_Queue1,true,false,false,null);
channel.queueDeclare(Constant.Pubsub_Queue2,true,false,false,null);
//5.绑定交换机与队列
channel.queueBind(Constant.Pubsub_Queue1,Constant.Exchange,"");
channel.queueBind(Constant.Pubsub_Queue2,Constant.Exchange,"");
//5.发送消息
for(int i=0;i<10;i++) {
String msg = "hello,pusub-Queue"+i;
channel.basicPublish(Constant.Exchange, "", null, msg.getBytes());
}

消费者

多个消费者，各自声明一个队列，他们所有的队列都被绑定在生产者声明的交换机上，消费者就可以监听到消息，从而进行处理

//3.声明交换机
channel.exchangeDeclare(Constant.Exchange, BuiltinExchangeType.FANOUT,true);
//4.声明队列
channel.queueDeclare(Constant.Pubsub_Queue2,true,false,false,null);
//5.消费消息
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
//TODO
System.out.println(new String(body));
}
};
channel.basicConsume(Constant.Pubsub_Queue2,true,consumer);

在SpringBoot中

声明交换机，队列和绑定关系

//发布订阅模式
@Bean("fanout_Queue1")
public Queue fountQueue1(){
return QueueBuilder.durable(Constant.FANOUT_QUEUE1).build();
}
@Bean("fanout_Queue2")
public Queue fountQueue2(){
return QueueBuilder.durable(Constant.FANOUT_QUEUE2).build();
}
@Bean("fanout_Exchange")
public FanoutExchange fanoutExchange(){
return ExchangeBuilder.fanoutExchange(Constant.FANOUT_EXCHANGE).durable(true).build();
}
@Bean
public Binding fanoutBinding1(@Qualifier("fanout_Exchange") FanoutExchange fanoutExchange,
@Qualifier("fanout_Queue1") Queue queue){
return BindingBuilder.bind(queue).to(fanoutExchange);
}
@Bean
public Binding fanoutBinding2(@Qualifier("fanout_Exchange") FanoutExchange fanoutExchange,
@Qualifier("fanout_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(fanoutExchange);
}

消费者

通过RabbitListener监听队列，接收参数 Message  String  Channel 三种类型

处理消息体

@Component
public class FanoutListener {
@RabbitListener(queues = Constant.FANOUT_QUEUE1)
public void fountListener1(Message message){
System.out.println("队列1"+"接收到消息"+"["+Constant.FANOUT_QUEUE1+"]"+message);
}
@RabbitListener(queues = Constant.FANOUT_QUEUE2)
public void fountListener2(Message message){
System.out.println("队列2"+"接收到消息"+"["+Constant.FANOUT_QUEUE2+"]"+message);
}
}

生产者

@RequestMapping("/fanout")
public String fountRabbit(){
for (int i = 0; i < 10; i++) {
String msg = "hello fount queue …."+i;
rabbitTemplate.convertAndSend(Constant.FANOUT_EXCHANGE,"", msg);
}
return "消息 发送成功";
}

4.路由模式

是发布订阅模式的变种，在其基础上，增加路由key

发布订阅模式是无条件的将所有消息分发给所有消费者，路由模式是交换机根据RountingKey的规则，将数据筛选后，发给对应的消费者队列

使用场景：需要根据特定规则分发消息的场景

• 日志分级处理：error 日志发送到报警队列，info/warning 日志发送到存档队列。

• 按业务类型路由消息（如订单支付成功、订单取消分别路由到不同队列）

县借助普通Java代码理解关系

//1.建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost(Constant.VirtualHost);
connectionFactory.setHost(Constant.Host);
connectionFactory.setPort(Constant.Port);
connectionFactory.setUsername(Constant.Username);
connectionFactory.setPassword(Constant.Password);
Connection connection=connectionFactory.newConnection();
//2.创建信道
Channel channel=connection.createChannel();
//3.声明交换机
channel.exchangeDeclare(Constant.Direct_Exchange, BuiltinExchangeType.DIRECT,true);
//4.声明队列
channel.queueDeclare(Constant.Direct_Queue1,true,false,false,null);
channel.queueDeclare(Constant.Direct_Queue2,true,false,false,null);
//5.绑定交换机与队列
channel.queueBind(Constant.Direct_Queue1,Constant.Direct_Exchange,"a");
channel.queueBind(Constant.Direct_Queue2,Constant.Direct_Exchange,"a");
channel.queueBind(Constant.Direct_Queue2,Constant.Direct_Exchange,"b");
channel.queueBind(Constant.Direct_Queue2,Constant.Direct_Exchange,"c");
//6.发送消息
String msg="hello direct…….is a";
channel.basicPublish(Constant.Direct_Exchange,"a",null,msg.getBytes());
String msg1="hello direct…….is b";
channel.basicPublish(Constant.Direct_Exchange,"b",null,msg1.getBytes());
String msg2="hello direct…….is c";
channel.basicPublish(Constant.Direct_Exchange,"c",null,msg2.getBytes());
//7.释放资源
System.out.println("消息发送完毕！");
channel.close();
connection.close();
}

在springBoot中 声明队列，交换机与绑定关系

//路由模式
@Bean("direct_Exchange")
public DirectExchange directExchange(){
return ExchangeBuilder.directExchange(Constant.DIRECT_EXCHANGE).durable(true).build();
}
@Bean("direct_Queue1")
public Queue directQueue1(){
return QueueBuilder.durable(Constant.DIRECT_QUEUE1).build();
}
@Bean("direct_Queue2")
public Queue directQueue2(){
return QueueBuilder.durable(Constant.DIRECT_QUEUE2).build();
}
@Bean
public Binding directBinding1(@Qualifier("direct_Exchange") DirectExchange directExchange,
@Qualifier("direct_Queue1") Queue queue){
return BindingBuilder.bind(queue).to(directExchange).with("orange");
}
@Bean
public Binding directBinding2(@Qualifier("direct_Exchange") DirectExchange directExchange,
@Qualifier("direct_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(directExchange).with("orange");
}
@Bean
public Binding directBinding3(@Qualifier("direct_Exchange") DirectExchange directExchange,
@Qualifier("direct_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(directExchange).with("black");
}
@Bean
public Binding directBinding4(@Qualifier("direct_Exchange") DirectExchange directExchange,
@Qualifier("direct_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(directExchange).with("green");
}

在消费者中

@Component
public class DirectListener {
@RabbitListener(queues = Constant.DIRECT_QUEUE1)
public void directListener1(String message){
System.out.println("["+Constant.DIRECT_QUEUE1+"]"+"接收到消息"+message);
}

@RabbitListener(queues = Constant.DIRECT_QUEUE2)
public void directListener2(String message){
System.out.println("["+Constant.DIRECT_QUEUE2+"]"+"接收到消息"+message);
}
}

在生产者中

@RequestMapping("/direct/{routingKey}")
public String directRabbit(@PathVariable("routingKey") String routingKey){
rabbitTemplate.convertAndSend(Constant.DIRECT_EXCHANGE,routingKey,"hello,direct….."+routingKey);
return "消息发送成功";
}

5.通配符模式

路由模式的升级，BingdingKey支持通配符

• 复杂的多条件订阅，例如物联网设备上报数据按 设备类型.区域.级别 分发。

• 电商系统中按 商品类目.品牌.活动 路由促销消息

通配符中， *代表匹配一个单词，不能是零个，也不能是多个

                    #代表匹配零个或多个单词

                    .作为单词之间的分割，无匹配单词含义

普通代码示例生产者：

//1.建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost(Constant.VirtualHost);
connectionFactory.setHost(Constant.Host);
connectionFactory.setPort(Constant.Port);
connectionFactory.setUsername(Constant.Username);
connectionFactory.setPassword(Constant.Password);
Connection connection=connectionFactory.newConnection();
//2.创建信道
Channel channel=connection.createChannel();
//3.声明交换机
channel.exchangeDeclare(Constant.Topic_Exchange, BuiltinExchangeType.TOPIC,true);
//4.声明队列
channel.queueDeclare(Constant.Topic_Queue1,true,false,false,null);
channel.queueDeclare(Constant.Topic_Queue2,true,false,false,null);
//5.绑定交换机与队列
channel.queueBind(Constant.Topic_Queue1,Constant.Topic_Exchange,"*.a");
channel.queueBind(Constant.Topic_Queue1,Constant.Topic_Exchange,"*.b");
channel.queueBind(Constant.Topic_Queue2,Constant.Topic_Exchange,"a.#");
channel.queueBind(Constant.Topic_Queue2,Constant.Topic_Exchange,"b.c.*");
channel.queueBind(Constant.Topic_Queue2,Constant.Topic_Exchange,"c.b.*");
//6.发送消息
String msg="hello direct…….is c.a";
channel.basicPublish(Constant.Topic_Exchange,"c.a",null,msg.getBytes());
String msg1="hello direct…….is a.b";
channel.basicPublish(Constant.Topic_Exchange,"a.b",null,msg1.getBytes());
String msg4="hello direct…….is abcdefg";
channel.basicPublish(Constant.Topic_Exchange,"a.array",null,msg4.getBytes());
String msg2="hello direct…….is b.c.array";
channel.basicPublish(Constant.Topic_Exchange,"b.c.*",null,msg2.getBytes());
String msg3="hello direct…….is c.b.tmp";
channel.basicPublish(Constant.Topic_Exchange,"c.b.*",null,msg3.getBytes());
//7.释放资源
System.out.println("消息发送完毕！");
channel.close();
connection.close();

在springBoot中

声明队列 交换机，绑定关系设置routingKey

//通配符模式
@Bean("Topic_Exchange")
public TopicExchange topicExchange(){
return ExchangeBuilder.topicExchange(Constant.TOPIC_EXCHANGE).durable(true).build();
}
@Bean("Topic_Queue1")
public Queue topicQueue1(){
return QueueBuilder.durable(Constant.TOPIC_QUEUE1).build();
}
@Bean("Topic_Queue2")
public Queue topicQueue2(){
return QueueBuilder.durable(Constant.TOPIC_QUEUE2).build();
}
@Bean
public Binding bindingQueue1(@Qualifier("Topic_Exchange") TopicExchange topicExchange,
@Qualifier("Topic_Queue1") Queue queue){
return BindingBuilder.bind(queue).to(topicExchange).with("*.orange");
}
@Bean
public Binding bindingQueue2(@Qualifier("Topic_Exchange") TopicExchange topicExchange,
@Qualifier("Topic_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(topicExchange).with("*.orange");
}
@Bean
public Binding bindingQueue3(@Qualifier("Topic_Exchange") TopicExchange topicExchange,
@Qualifier("Topic_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(topicExchange).with("#.black");
}
@Bean
public Binding bindingQueue4(@Qualifier("Topic_Exchange") TopicExchange topicExchange,
@Qualifier("Topic_Queue2") Queue queue){
return BindingBuilder.bind(queue).to(topicExchange).with("*.green");
}

6.RPC模式

RabbitMQ 不仅可以做异步通知，也可以实现同步的请求/响应模式。客户端发送请求消息时，在消息属性中指定一个回调队列和关联 ID（correlationId），服务端处理后将结果发送到该回调队列。客户端通过 correlationId 匹配请求与响应

• 微服务间的同步调用（替代 HTTP，降低耦合）。

• 计算密集型任务的远程执行

客户端代码：

public class Client {
//做两件事情，发送请求携带replyTo，CorrelationID
//接收请求 校验CorrelationId
public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
//建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost(Constant.VirtualHost);
connectionFactory.setHost(Constant.Host);
connectionFactory.setPort(Constant.Port);
connectionFactory.setUsername(Constant.Username);
connectionFactory.setPassword(Constant.Password);
Connection connection=connectionFactory.newConnection();
//创建信道
Channel channel=connection.createChannel();
//声明队列
channel.queueDeclare(Constant.RPC_Request_Queue,true,false,false,null);
channel.queueDeclare(Constant.RPC_Response_Queue,true,false,false,null);
//发送消息
String msg="hello RPC…….!";
//replyTo
//CorrelationID
String CorrelationID= UUID.randomUUID().toString();
//设置属性
AMQP.BasicProperties props=new AMQP.BasicProperties().builder()
.replyTo(Constant.RPC_Request_Queue)
.correlationId(CorrelationID)
.build();
channel.basicPublish("",Constant.RPC_Request_Queue,props,msg.getBytes());
//接收响应
//需要一个阻塞队列 因为代码运行，如果不阻塞，他会直接运行完毕，此时消息还未经server发送
BlockingDeque<String>blockingDeque=new LinkedBlockingDeque<>(1);
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
String ResMsg=new String(body);
System.out.println("接收到回调信息："+ResMsg);
//校验响应信息
if(properties.getCorrelationId().equals(CorrelationID)){
//成功
blockingDeque.offer(ResMsg);
}
}
};
channel.basicConsume(Constant.RPC_Response_Queue,true,consumer);
String result=blockingDeque.take();
System.out.println("RPC 响应结果: "+result);
}
}

服务端代码

public class Server {
//做两件事：接收消息 ，发送响应
public static void main(String[] args) throws IOException, TimeoutException {
//建立连接
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setVirtualHost(Constant.VirtualHost);
connectionFactory.setHost(Constant.Host);
connectionFactory.setPort(Constant.Port);
connectionFactory.setUsername(Constant.Username);
connectionFactory.setPassword(Constant.Password);
Connection connection=connectionFactory.newConnection();
//创建信道
Channel channel=connection.createChannel();
//声明队列
channel.queueDeclare(Constant.RPC_Request_Queue,true,false,false,null);
channel.queueDeclare(Constant.RPC_Response_Queue,true,false,false,null);
//接收消息
//保存请求的信息 ReplyTo,CorrelationID
channel.basicQos(1);//高级特性后续补充
DefaultConsumer consumer=new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// super.handleDelivery(consumerTag, envelope, properties, body);
//TODO 此处简单处理
String resMsg=new String(body);
System.out.println("接收请求消息："+resMsg);
String response="处理请求："+resMsg;
//发送响应
String correlationId = properties.getCorrelationId();
AMQP.BasicProperties props=new AMQP.BasicProperties().builder()
.correlationId(correlationId)
.build();
System.out.println("发送响应："+response);
channel.basicPublish("",Constant.RPC_Response_Queue,props,response.getBytes(StandardCharsets.UTF_8));
channel.basicAck(envelope.getDeliveryTag(),false);//确认
}
};
channel.basicConsume(Constant.RPC_Request_Queue,false,consumer);//手动确认高级特性
}
}

7.发布确认模式

生产者将chanel设置为confirm模式，发布的每一条消息，都会获得一个唯一的ID，生产者可以将这些序列号与消息关联起来，以便跟踪消息的状态，当消息被RabbitMQ接收并处理后，服务器会异步的向生产者发送一个ACK（包含消息的唯一ID），表明消息已经到达

适用于金融交易或订单处理

channel.confirmSelect(); // 开启确认模式

channel.addConfirmListener((sequenceNumber, multiple) -> {
// ack 处理
System.out.println("Message confirmed: " + sequenceNumber);
}, (sequenceNumber, multiple) -> {
// nack 处理，可重发
System.err.println("Message nack-ed: " + sequenceNumber);
});

// 正常发送消息
channel.basicPublish(exchange, routingKey, null, body);

总结

RabbitMQ 的模式虽多，但核心是 Exchange 与 Queue 的绑定关系以及 Routing Key 的匹配规则。理解了这层抽象，再复杂的拓扑也能化繁为简