当前位置： 首页 > news >正文

Rabbit MQ的消息模式-Java原生代码

news 2025/8/2 0:23:17

一.简单模式

1.1.核心逻辑

生产者 → 队列 → 单个消费者（1:1 直连），消息被消费后自动从队列删除。

1.2.关键特性

  • 无交换器（其实使用的是默认交换机不是显示指定），直接指定队列 
  • 消息默认自动确认（autoAck），易丢失消息 

1.3.应用场景

单任务即时处理（如聊天消息、简单日志）

1.4.架构图

1.5.代码示例

        在Rabbit中，生产者发送完消息后，就结束了，之后的操作就与生产者无关了，而消费者是被动接收的，一直处于监听状态。 

  • pom依赖
        <dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.20.0</version></dependency>
  • 生产者 
package com.example.demo.rabbitmq.simple;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {public static void main(String[] args)throws Exception {// 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置主机地址connectionFactory.setHost("127.0.0.1");//设置连接端口号:默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称:默认为/connectionFactory.setVirtualHost("/");//设置连接用户名;默认为guestconnectionFactory.setUsername("guest");//设置连接密码;默认为guestconnectionFactory.setPassword("guest");//1 创建连接Connection connection = connectionFactory.newConnection();//创建频道Channel channel=connection.createChannel();/*** 声明(创建)队列* 如果没有一个名字叫simp1e-queue的队列，则会创建该队列，如果有则不会创建,所以该方法在确认有消息该消息队列的情况下可以省略* queue      参数1:队列名称* durable    参数2:是否定义持久化队列，当MQ重启之后还在* exclusive  参数3:是否独占本次连接。若独占，只能有一个消费者监听这个队列目Connection关闭时删除这个队列* autoDelete 参数4:是否在不使用的时候自动删除队列，也就是在没有Consumer时自动删除* arquments  参数5:队列其它参数*/channel.queueDeclare("simple_queue", true, false, false, null);//要发送的信息String message="Hello RabbitMQ!";/*** 指定消息队列* 参数1:交换机名称,如果没有指定则使用默认Default_Exchange* 参数2:路由key,简单模式可以传递队列名称* 参数3:配置信息* 参数4:消息内容*/channel.basicPublish( "","simple_queue",  null, message.getBytes());channel.close();connection.close();}
}

 生产者main方法执行日志

        如果你先启动生产者main方法，那么你可以在RabbitMQ的web页面可以看到在队列中有一条消息。如果消费者一直都是监听的，那大概率看不到，因为生产者发送消息的那一刻立马就被消费者接收了，在消息队列中就删除了。

 

  • 消费者
package com.example.demo.rabbitmq.simple;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class Consumer {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2.设置参数factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//3.创建连接 ConnectionConnection connection = factory.newConnection();//4.创建channe1Channel channel = connection.createChannel();/***5.创建队列* 如果没有一个名字叫simp1e-queue的队列，则会创建该队列，如果有则不会创建,所以该方法在确认有消息该消息队列的情况下可以省略* 数1.queue:队列名称* 参数2.durab1e:是否持久化。如果持久化，则当MQ重启之后还在* 参数3.exclusive:是否独占。* 参数4.autoDelete:是否自动删除。当没有Consumer时，自动删除掉* 参数5.arguments:队列其它参数*/channel.queueDeclare("simple_queue", true, false, false, null);// 接收消息DefaultConsumer consumer=new DefaultConsumer(channel){/*** 接收到消息后，此方法将被调用* @param consumerTag 标识* @param envelope 获取一些信息，交换机，路由key...* @param properties 配置信息* @param body 数据* @throws IOException*/@Overridepublic void handleDelivery(String consumerTag, Envelope envelope,AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("consumerTag:"+consumerTag);System.out.println("Exchange:"+envelope.getExchange());System.out.println("RoutingKey:"+envelope.getRoutingKey());System.out.println("properties:"+properties);System.out.println("body:"+new String(body));}};// 参数1.queue:队列名称// 参数2.autoAck:是否自动确认，类似咱们发短信，发送成功会收到一个确认消息// 参数3.callback:回调对象// 消费者类似一个监听程序，主要是用来监听消息channel.basicConsume("simple_queue",true,consumer);}
}

 接收后，消息队列就是空的了。

你也可以点击上面图片中消息队列的名称simple_queue，查看下面具体详情，如下面图片。 

 

在查看消息时，注意在右上角选择页面的刷新频率。 

   

二.工作模式

2.1.核心逻辑

生产者 → 队列 → 多个消费者并行消费（1:N）。

工作模式与简单模式唯一的不同在于它有多个消费者，当队列中有消息时，多个消费者竞争，每条消息仅被一个消费者处理。

2.2.关键特性

  • 多个消费者竞争消费同一队列，默认轮询分发（Round-Robin）
  • 可配置 basicQos(prefetchCount) 实现公平分发（能者多劳）

2.3.应用场景

资源密集型任务并行处理（如文件转码、批量邮件）

2.4.架构图

2.5.代码示例

main方法启动消费者1和消费者2，当然你可以多创建几个消费者，复制简单模式中的代码即可。

修改下生产者代码，创建一个新的消息队列，并且发送10条消息

消费者1接收到的信息

 

消费者2接收到的消息

 

三.发布订阅模式

交换器类型：fanout（广播）

3.1.核心逻辑

生产者 → Fanout交换器 → 绑定队列 → 所有消费者
忽略路由键（Routing Key） ，消息复制到所有绑定队列。

3.3.关键特性

  • 一条消息被多个消费者独立消费（广播）
  • 需显式绑定队列到交换器 

3.3.应用场景

事件广播（如用户注册后同时发邮件、短信）

3.4.架构图

 如上图所示，发布订阅模式有以下特点;

  • 指定类型的交换机；
  • 多个消息队列，交换机会将一条消息发布到每一个消息队列中；
  • 每个消息队列可以有一个或者多个消费者；

3.5.代码示例

生产者代码

package com.example.demo.rabbitmq.fanout;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {private static final String EXCHANGE_NAME="test_fanout_exchange";public static void main(String[] args)throws Exception {// 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置主机地址connectionFactory.setHost("127.0.0.1");//设置连接端口号:默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称:默认为/connectionFactory.setVirtualHost("/");//设置连接用户名;默认为guestconnectionFactory.setUsername("guest");//设置连接密码;默认为guestconnectionFactory.setPassword("guest");//1 创建连接Connection connection = connectionFactory.newConnection();//创建频道Channel channel=connection.createChannel();/*** 创建交换机* 参数1:交换机名称* 参数2:交换机类型* 参数3.durable:是否持久化* 参数4.autoDelete:自动删除* 参数5.internal:内部使用，一般false* 参数6.arquments:其它参数*///这两个方法是一样的//channel.exchangeDeclare(EXCHANGE_NAME, "fanout");channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT, true, false, false, null);/*** 声明(创建)队列* 如果没有一个名字叫simp1e-queue的队列，则会创建该队列，如果有则不会创建,所以该方法在确认有消息该消息队列的情况下可以省略* queue      参数1:队列名称* durable    参数2:是否定义持久化队列，当MQ重启之后还在* exclusive  参数3:是否独占本次连接。若独占，只能有一个消费者监听这个队列目Connection关闭时删除这个队列* autoDelete 参数4:是否在不使用的时候自动删除队列，也就是在没有Consumer时自动删除* arquments  参数5:队列其它参数*/channel.queueDeclare("fanout_queue_1", true, false, false, null);channel.queueDeclare("fanout_queue_2", true, false, false, null);channel.queueDeclare("fanout_queue_3", true, false, false, null);/*** 绑定队列到交换机* 参数1:队列名称* 参数2:交换机名称* 参数3:路由key 交换机的类型为fanout，为空*/channel.queueBind("fanout_queue_1", EXCHANGE_NAME, "");channel.queueBind("fanout_queue_2", EXCHANGE_NAME, "");channel.queueBind("fanout_queue_3", EXCHANGE_NAME, "");//要发送的信息String message="Hello RabbitMQ!";/*** 指定消息队列* 参数1:交换机名称,如果没有指定则使用默认Default_Exchange* 参数2:路由key,简单模式可以传递队列名称* 参数3:配置信息* 参数4:消息内容*/channel.basicPublish( EXCHANGE_NAME,"",  null, message.getBytes());channel.close();connection.close();}
}

上面代码有以下作用：

  • 创建指定类型的交换机(如果有，不创建)；
  • 创建三个消息队列(如果有，不创建)；
  • 绑定交换机与消息队列；
  • 发送消息； 

运行后可在Rabbit MQ的管理页面查看到下面的内容

交换机

点击交换机名称，查看绑定关系，绑定了三个消息队列 

 

点击任何一个消息队列，你都可以看到有一条消息

 

或者你可以根据下图查看消息队列中的消息

 

如果你要查看某个队列中的具体消息，点击队列名称，找到Get messages。

如果不修改图中的任何一个选项，是不会删除队列中的消息的。者三个消息队列中，都有一条消息【Hello RabbitMQ!】这就是广播的效果

 

消费端代码 

消费端代码并没有变化，与简单模式和工作模式的没有不同，因为消费端监听的是消息队列，只需要修改消息队列名称后运行即可。

 

在上面图片中，我只创建了两个消费者，分别监听了队列1和队列2。通过下面可以看到，队列1和队列2的消息被接收了。如果你感兴趣，可以多加几个消费者，注意：一个队列可以有多个消费者。

 

四.路由模式

交换器类型：direct（精确匹配）

4.1.核心逻辑

生产者 → Direct交换器 → 匹配路由键的队列 → 消费者
路由键需与绑定键（Binding Key）完全一致。

4.2.关键特性

  • 实现消息分类投递（如按日志级别分发）
  • 队列可绑定多个路由键 

4.3.应用场景

精准路由（如ERROR日志存数据库，INFO日志打印）

4.4.架构图

根据上图，路由模式有以下特点

  • 特定类型的交换机 direct；
  • 指定具体的路由，交换机根据路由将消息发送到对应的队列中；
  • 需要注意的是，交换机到队列的路由规则，可以多个。

4.5.代码示例

消费者

消费端的代码没什么不同，在这里，我创建了两个消费端，各监听一个消息队列。

生产者

package com.example.demo.rabbitmq.routing;import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class Producer {private static final String EXCHANGE_NAME="direct_exchange";public static void main(String[] args)throws Exception {// 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置主机地址connectionFactory.setHost("127.0.0.1");//设置连接端口号:默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称:默认为/connectionFactory.setVirtualHost("/");//设置连接用户名;默认为guestconnectionFactory.setUsername("guest");//设置连接密码;默认为guestconnectionFactory.setPassword("guest");//1 创建连接Connection connection = connectionFactory.newConnection();//创建频道Channel channel=connection.createChannel();/*** 创建交换机* 参数1:交换机名称* 参数2:交换机类型* 参数3.durable:是否持久化* 参数4.autoDelete:自动删除* 参数5.internal:内部使用，一般false* 参数6.arquments:其它参数*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT, true, false, false, null);/*** 声明(创建)队列* 如果没有一个名字叫simp1e-queue的队列，则会创建该队列，如果有则不会创建,所以该方法在确认有消息该消息队列的情况下可以省略* queue      参数1:队列名称* durable    参数2:是否定义持久化队列，当MQ重启之后还在* exclusive  参数3:是否独占本次连接。若独占，只能有一个消费者监听这个队列目Connection关闭时删除这个队列* autoDelete 参数4:是否在不使用的时候自动删除队列，也就是在没有Consumer时自动删除* arquments  参数5:队列其它参数*/channel.queueDeclare("direct_queue_1", true, false, false, null);channel.queueDeclare("direct_queue_2", true, false, false, null);/*** 绑定队列到交换机* 参数1:队列名称* 参数2:交换机名称* 参数3:路由key*/channel.queueBind("direct_queue_1", EXCHANGE_NAME, "error");channel.queueBind("direct_queue_2", EXCHANGE_NAME, "info");//要发送的信息String message="日志信息:张三调用了delete方法.错误了，目志级别error";/*** 指定消息队列* 参数1:交换机名称,如果没有指定则使用默认Default_Exchange* 参数2:路由key,简单模式可以传递队列名称* 参数3:配置信息* 参数4:消息内容*/channel.basicPublish( EXCHANGE_NAME,"error",  null, message.getBytes());channel.close();connection.close();}
}

 在上面代码中，重点在于：

  • 创建了一个交换机，类型为DIRECT；
  • 创建了两个队列；
  • 绑定交换机与队列的关系，并指定路由；发送消息时需指定交换机名称和路由key;

运行上面代码后，消费者1接收到了error消息。

修改生产者代码中的路由key，再次执行 

消费者2接收到了信息

 

五.主题模式

交换器类型：topic（模糊匹配）

5.1.核心逻辑

生产者 → Topic交换器 → 通配符匹配的队列 → 消费者
路由键支持 *（匹配一词）和 #（匹配多词），如 user.*.order。

5.2.关键特性

  • 动态路由（如按用户兴趣订阅消息）
  • 绑定键格式示例：news.#（接收所有新闻）

5.3.应用场景

动态消息分发（如电商系统按用户标签推送促销）

5.4.架构图


主题模式和路由模式的区别在于，主题模式的路由key可以模糊匹配 。

将交换机的类型设置为topic类型，在绑定队列时配置路由key，可以设置模糊匹配的规则，如下图

在生产者发送消息后，交换机根据路由开始匹配，将消息发送到所有匹配的队列中。

六.头模式

交换器类型：headers（键值对匹配）

6.1.核心逻辑

生产者 → Headers交换器 → 匹配消息头的队列 → 消费者
通过 x-match 指定 all（全匹配）或 any（任一匹配）。

6.2.关键特性

  • 不依赖路由键，用消息头（Headers）路由 
  • 性能较低，极少使用 

6.3.应用场景

特殊路由需求（如按消息语言或版本过滤）

6.4.架构图

        该模式与上文中的各个模式完全不同，在头模式中，生产者不指定或创建消息队列，不绑定交换机与消息队列。这部分功能在消费者中。

在生产者中，有以下几步：

  • 声明或创建交换机；
  • 构建消息属性，指定消息头；

6.5.代码案例

消费者

在该模式中，必须先启动消费者，因为生产者没有指定消息队列，如果先启动生产者，会导致数据丢失。

package com.example.demo.rabbitmq.head;import com.rabbitmq.client.*;import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1.创建连接工厂ConnectionFactory factory = new ConnectionFactory();//2.设置参数factory.setHost("127.0.0.1");factory.setPort(5672);factory.setVirtualHost("/");factory.setUsername("guest");factory.setPassword("guest");//3.创建连接 ConnectionConnection connection = factory.newConnection();//4.创建channe1Channel channel = connection.createChannel();//5.创建交换机channel.exchangeDeclare("headers_exchange", BuiltinExchangeType.HEADERS, true, false, false, null);/***5.创建队列* 如果没有一个名字叫simp1e-queue的队列，则会创建该队列，如果有则不会创建,所以该方法在确认有消息该消息队列的情况下可以省略* 数1.queue:队列名称* 参数2.durab1e:是否持久化。如果持久化，则当MQ重启之后还在* 参数3.exclusive:是否独占。* 参数4.autoDelete:是否自动删除。当没有Consumer时，自动删除掉* 参数5.arguments:队列其它参数*/channel.queueDeclare("headers_queue_1", true, false, false, null);// 设置绑定参数（完全匹配）Map<String, Object> bindingArgs = new HashMap<>();bindingArgs.put("x-match", "all"); // 必须所有Header匹配bindingArgs.put("format", "JSON");bindingArgs.put("priority", "high");/*** 绑定队列到交换机* 参数1:队列名称* 参数2:交换机名称* 参数3:路由key 为空* 参数4:绑定参数*/channel.queueBind("headers_queue_1", "headers_exchange", "", bindingArgs);// 接收消息// 消费消息DeliverCallback deliverCallback = (consumerTag, delivery) -> {String message = new String(delivery.getBody(), "UTF-8");System.out.println(" [x] Received (ALL match): " + message +" Headers: " + delivery.getProperties().getHeaders());};/*** 监听消息*  参数1.queue:队列名称*  参数2.autoAck:是否自动确认，类似咱们发短信，发送成功会收到一个确认消息*  参数3.callback:回调对象*  参数4.cancelCallback:取消消费的回调*  参数5.arguments:消费者其它参数*/channel.basicConsume("headers_queue_1",true,deliverCallback,consumerTag->{});}
}

生产者

package com.example.demo.rabbitmq.head;import com.rabbitmq.client.*;import java.util.HashMap;
import java.util.Map;public class Producer {private static final String EXCHANGE_NAME="headers_exchange";public static void main(String[] args)throws Exception {// 创建连接工厂ConnectionFactory connectionFactory = new ConnectionFactory();//设置主机地址connectionFactory.setHost("127.0.0.1");//设置连接端口号:默认为 5672connectionFactory.setPort(5672);// 虚拟主机名称:默认为/connectionFactory.setVirtualHost("/");//设置连接用户名;默认为guestconnectionFactory.setUsername("guest");//设置连接密码;默认为guestconnectionFactory.setPassword("guest");//1 创建连接Connection connection = connectionFactory.newConnection();//创建频道Channel channel=connection.createChannel();/*** 创建交换机* 参数1:交换机名称* 参数2:交换机类型* 参数3.durable:是否持久化* 参数4.autoDelete:自动删除* 参数5.internal:内部使用，一般false* 参数6.arquments:其它参数*/channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.HEADERS, true, false, false, null);//设置头消息Map<String, Object> headers = new HashMap<>();headers.put("format", "JSON");headers.put("priority", "high");// 构建消息属性AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().headers(headers) //设置头消息.build();String message = "Header Exchange Test Message";/*** 指定消息队列* 参数1:交换机名称,如果没有指定则使用默认Default_Exchange* 参数2:路由key,简单模式可以传递队列名称* 参数3:配置信息* 参数4:消息内容*/channel.basicPublish( EXCHANGE_NAME,"",  properties, message.getBytes());channel.close();connection.close();}
}

 消费者接收到的消息

当然Rabbit MQ还有其它模式，如，RPC模式：远程过程调用，本质上是同步调用，和我们使用OpenFeign调用远程接口一样，有机会再说。

http://www.lryc.cn/news/604793.html

相关文章：

  • YOLO-01目标检测基础
  • 02 基于sklearn的机械学习-特征降维（特征选择、PCA）、KNN算法、模型选择与调优（交叉验证、朴素贝叶斯算法、拉普拉斯平滑）
  • Android调用python库和方法的实现
  • YOLOv5u：无锚点检测的革命性进步
  • android-PMS-创建新用户流程
  • 舆情监测专员需要哪些常用软件工具?
  • 基于 Hadoop 生态圈的数据仓库实践 —— OLAP 与数据可视化（一）
  • 论文Review 3DGSSLAM S3PO-GS | ICCV 2025 港科广出品！| 高效快速的3DGSSLAM！
  • sqli-labs：Less-1关卡详细解析
  • CMS框架漏洞
  • 3D Web轻量化引擎HOOPS Communicator数据处理与流式加载能力概述
  • 【音视频】WebRTC-Web 音视频采集与播放
  • 【预判一手面试问题：排序】
  • 依托客户满意度分析协助企业精准把握市场趋势​（满意度调查）
  • 智能AI医疗物资/耗材管理系统升级改造方案分析
  • InfluxDB 与 Java 框架集成：Spring Boot 实战（二）
  • VSCode插件开发完整教程：从零开始创建文件导出插件
  • Python 程序设计讲义（37）：字符串的处理方法——设置字符串居中显示：center() 方法
  • 图像平滑处理
  • 9.项目起步（3）
  • OpenCV学习day1
  • 实习小记（个人中心的编辑模块）
  • 商标注册后可以随意更改字体和颜色吗！
  • 怎么理解锁相环主时钟（PLL）怎么做到对时钟进行倍频？
  • Keil STM32工程各文件作用
  • AI框架工具FastRTC快速上手2——整体框架及Stream类详解
  • 方块世界：失落文明的遗产
  • Deforum Stable Diffusion，轻松实现AI视频生成自由！
  • 语音识别dolphin 学习笔记
  • UE5多人MOBA+GAS 番外篇：将冷却缩减属性应用到技能冷却中