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

RabbitMQ详情介绍—七种工作模式

news 2025/8/10 7:33:47

目录

一、什么是消息队列？

系统之间的调用通常有两种方式：

二、MQ的作用（特点）

三、RabbitMQ的基本结构

四、RabbitMQ的工作流程

五、RabbitMQ工作模式介绍：

1、Simple（简单模式）

实现步骤：

引入依赖：

编写生产者代码：

编写消费者代码：

2、Work Queue（工作队列）

实现步骤：

引入依赖：

编写生产者代码：

编写消费者代码：

3、Publish/Subscribe(发布/订阅)

 实现步骤：

引入依赖：

编写生产者代码：

编写消费者代码：

观察结果：

4、Routing（路由模式）

 实现步骤：

引入依赖：

编写生产者代码：

编写消费者代码：

观察结果：

5、Topics(通配符模式)

 实现步骤：

引入依赖：

编写生产者代码：

编写消费者代码：

观察结果：

6、RPC（RPC通信）

 实现步骤：

引入依赖：

编写客户端代码：

编写RPC server代码：

7、Publisher Confirms(发布确认）

一、什么是消息队列？

消息队列（Message Queue）是一种在分布式系统中实现异步通信的中间件技术，它通过存储和传递消息（数据），实现不同组件、服务或系统之间的解耦和高效协作。简单来说，消息队列就像一个 “中转站” 或 “邮箱”，发送方（生产者）将消息放入队列，接收方（消费者）从队列中取出消息并处理，双方无需直接交互，甚至可以在不同时间、不同状态下工作

MabbitMQ：是一款基于 AMQP（Advanced Message Queuing Protocol）协议的开源消息队列中间件，由 Erlang 语言开发，以轻量级、高可靠性、灵活的路由机制和丰富的消息模式著称。它广泛应用于分布式系统中，实现服务间的异步通信、解耦和流量控制。

RabbitMQ介绍官⽹: RabbitMQ: One broker to queue them all | RabbitMQ

系统之间的调用通常有两种方式：

  • 同步通信：

直接调用对方的服务，数据从一端直接到达另一端

  • 异步通信：
数据从⼀端发出后，先进⼊⼀个容器进⾏临时存储，当达到某种条件后，再由这个容器发送给另⼀端. 容器的⼀个具体实现就是MQ( message queue 

二、MQ的作用（特点）

1、解耦

通过引用消息队列进行通信，打破服务间的直接依赖，生产者只需要将消息发送到队列，无需关心消费者的具体实现（无需直接调用对方接口方法）。当消费者逻辑改变或者新增消费方式时，无需修改生产者代码。使得它们可以独立演化和扩展，大大降低系统耦合度，

2、流量削峰

消息队列可以作为一个缓冲区，当系统面临突然高并发，消息队列可暂存大量请求，而消费者按自身处理能力逐步消费，避免系统被突然的高负载压垮

例如：双十一秒杀活动，瞬间涌入数十万订单请求进入消息队列，后端服务每秒处理1000单，平稳消化流量

3、异步通信

生产者发送消息后无需等待消费者处理结果，可立即返回，提高系统的并发性和响应速度。避免同步调用中的“串行等待”问题

4、延迟通知

延迟通知是指在特定时间点或经过指定延迟时间后，自动触发的消息通知机制。在实际业务中应用广泛，例如订单超时未支付自动取消通知、会议开始前 30 分钟提醒。（具体实现是利用死信队列机制和消息过期时间TTl）

三、RabbitMQ的基本结构

名称描述
Producer（生产者）是RabbitMQ Server的客户端，用来向消息队列发送消息
Consumer（消费者）是RabbitMQ Server的客户端，用来向消息队列接收消息
Connection（连接）连接是生产者和消费者与RabbitMQ之间的连接。每个生产者和消费者都需要与RabbitMQ建立一个连接，以便发送和接收消息。连接通常是长连接，可以重用以提高性能和效率。
Channel（信道）Channel是连接（Connection）内的逻辑通道，用于完成大部分 AMQP 操作，如声明队列、发送和接收消息等。在 RabbitMQ 中引入 Channel（信道）的主要目的是为了提高系统的性能、灵活性和效率。使用 Channel 可以避免频繁地创建和销毁连接，因为一个连接可以包含多个 Channel。这样可以减少连接的开销，节省系统资源，并提高性能。
Exchange（交换机）

交换机是消息的接收和收发中心，负责接收生产者发送的消息，并根据指定的路由规则发送到一个或多个队列中。（Exchange相当于Queue的代理，可以设置不同的写入策略，写入到对应的队列中。对于队列的写入，更加灵活）

交换机的类型有：Fanout扇出、Topic主题、Direct直出

Queue（队列）队列是消息的缓存区，用于存储交换机发送的消息。生产者发送的消息最终会被存储在队列中，等待消费者进行消费。队列可以持久化到磁盘，以确保消息不会在RabbitMQ宕机或重启后丢失。
Virtual host（虚拟主机）
这是⼀个虚拟概念. 它为消息队列提供了⼀种逻辑上的隔离机制. 对于 RabbitMQ⽽⾔, ⼀个 BrokerServer 上可以存在多个 Virtual Host. 当多个不同的⽤⼾使⽤同⼀个 RabbitMQ Server 提供的服务时，可以虚拟划分出多个 vhost，每个⽤⼾在⾃⼰的 vhost 创建 exchange/queue 等
类似MySQL的"database", 是⼀个逻辑上的集合. ⼀个MySQL服务器可以有多个database
Broker（RabbitMQ服务器）

是RabbitMQ Server,主要是接收和收发消息

四、RabbitMQ的工作流程

  1. Producer 准备阶段

    • Producer 生产消息时，会为消息设置基本属性（如消息 ID、持久化标记delivery_mode、过期时间expiration、路由键routing_key等），这些属性会影响消息的路由、存储和处理逻辑。

  2. 连接与信道建立

    • Producer 与 RabbitMQ Broker 建立 TCP 连接（Connection），为了减少 TCP 连接开销，通常在连接内创建信道（Channel），后续所有操作（声明交换机、发送消息等）均通过信道完成（信道是轻量级的连接，共享 TCP 连接资源）。

  3. 交换机与队列的声明与绑定

    • 声明交换机（Exchange）：Producer 需指定交换机的类型（如direct、topic、fanout、headers），不同类型的交换机决定了消息的路由规则（这一步可省略，若未声明则使用默认交换机(AMQP default)）。
    • 声明队列（Queue）：Producer 或 Consumer 需声明队列，指定队列属性（如是否持久化durable、是否排他exclusive、是否自动删除auto_delete、死信参数等）。
    • 绑定交换机与队列：必须通过 绑定（Binding） 将交换机与队列关联，并指定绑定键（binding_key）。交换机接收消息后，会根据自身类型、消息的routing_key与队列的binding_key的匹配规则，决定将消息路由到哪些队列（这是你之前描述中缺失的关键环节：交换机不会直接将消息存入队列，而是通过绑定关系和路由规则转发）。

  4. Producer 发送消息

    • Producer 通过信道将消息发送至指定交换机，并指定消息的routing_key。
    • 若未指定交换机，则消息会发送到默认交换机（默认交换机是direct类型，会将消息路由到routing_key与队列名完全匹配的队列）。

  5. Broker 处理消息

    • 交换机根据自身类型和路由规则，将消息转发到匹配的队列：
      • 若找到匹配的队列，消息会被存入队列（队列会根据自身持久化配置决定是否写入磁盘）。
      • 若未找到匹配的队列，则根据 Producer 发送消息时设置的mandatory参数处理：
        • mandatory=True：Broker 会将消息退回给 Producer（通过basic.return方法）。
        • mandatory=False：Broker 直接丢弃消息。
    • 特殊情况：若消息设置了expiration（过期时间），且在队列中超过该时间未被消费，则会成为死信（Dead Letter），进入预先配置的死信队列（若未配置则丢弃）。

  6. Consumer 接收与处理消息

    • Consumer 同样通过 TCP 连接和信道连接到 Broker，并订阅目标队列（通过basic.consume方法）。
    • Broker 会将队列中的消息推送给消费者（或消费者主动拉取，如basic.get），推送策略可通过prefetch_count控制（限制消费者同时处理的未确认消息数量，避免过载）。
    • Consumer 处理消息后，需向 Broker 发送确认信号（Ack）：
      • 若发送basic.ack：Broker 从队列中删除该消息。
      • 若发送basic.nack或basic.reject（且未设置requeue=False）：消息会重新回到队列，等待再次消费。
      • 若 Consumer 崩溃且未发送 Ack：Broker 会认为消息未被处理，将其重新放入队列，确保消息不丢失（需队列持久化配合）。

五、RabbitMQ工作模式介绍：

1、Simple（简单模式）

图中P代表生产者，C代表消费者，Queue是队列名称。
⽣产者向其中投递消息, 消费者从其中取出消息
我们看到是没有Exchange的，但是RabbitMQ也会有一个默认的交换机。

这个默认的交换机通常被称为"amq.default"，是RabbitMQ自动创建的，用于在没有指定交换机的情况

实现步骤：

  1. 添加依赖
  2. 编写生产者代码
  3. 编写消费者代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写生产者代码：
/*** 简单模式：生产者* @date 2025/8/1*/
public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);  //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME); //账号connectionFactory.setPassword(Constant.PASSWORD); //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启信道Channel channel = connection.createChannel();//3.声明交换机 （此处使用内置交换机）//4.声明队列/*** queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)*  参数说明:*  queue: 队列名称*  durable: 可持久化*  exclusive: 是否独占*  autoDelete: 是否自动删除*  arguments: 参数*/channel.queueDeclare("hello",true,false,true,null);//5.发送消息/*** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* 参数说明:* exchange: 交换机名称* routingKey: 内置交换机, routingkey和队列名称保持一致* props: 属性配置* body: 消息*/for(int i = 0;i<10;i++){String msg = "hello world——"+i;channel.basicPublish("","hello",null,msg.getBytes());}System.out.println("发送消息成功");//6.资源释放}}
编写消费者代码：
public class ConsumerDemo {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);  //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME); //账号connectionFactory.setPassword(Constant.PASSWORD); //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启甬道Channel channel = connection.createChannel();//3.声明队列channel.queueDeclare("hello", true, false, true, null);//4.消费消息/*** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数说明:* queue: 队列名称* autoAck: 是否自动确认* callback: 接收到消息后, 执行的逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者接收到消息:" + new String(body));}};channel.basicConsume("hello",true,consumer);System.out.println("消费者启动成功");//5.资源释放}
}

2、Work Queue（工作队列）

一个生产者P，多个消费者C1、C2，在多个消息的情况下，队列会将消息分派给不用的消费者，每个消费者都会接收到不同的消息

特点：消息不会重复，分配给不同的消费者

实现步骤：

  1. 添加依赖
  2. 编写生产者代码
  3. 编写消费者代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写生产者代码：

工作队列模式和简单模式区别是有多个消费者，所以代码区别不大

相比简单模式，生产者的代码一样，为了清楚看到多个消费者竞争的关系，我们一次发送10天消息

//相关配置信息
public class Constant {public static final String HOST = "8.140.60.17";public static final int PORT = 5672;public static final String USER_NAME = "admin";public static final String PASSWORD = "admin";public static final String VIRTUAL_HOST = "xibbei";/*** 工作队列常量*/public static final String WORK_QUEUE = "work_queue";
}/*** 工作队列——生产者* 特点：消息不会重复，分配给不同的消费者，* 适用场景：集群环境中做异步处理*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);  //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME); //账号connectionFactory.setPassword(Constant.PASSWORD); //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel channel = connection.createChannel();//3.声明交换机 (使用内置交换机）//4.声明队列//如果队列不存在, 则创建, 如果队列存在, 则不创建channel.queueDeclare(Constant.WORK_QUEUE, true, false, true, null);//5.发送消息for(int i = 0;i<10;i++){String msg = "hello world——"+i;channel.basicPublish("", Constant.WORK_QUEUE, null, msg.getBytes());}System.out.println("发送消息成功");//6.资源释放}}
编写消费者代码：

工作队列需要两个消费者，两个消费者代码是相同的 

public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);  //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME); //账号connectionFactory.setPassword(Constant.PASSWORD); //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启甬道Channel channel = connection.createChannel();//3.声明队列channel.queueDeclare(Constant.WORK_QUEUE, true, false, true, null);//4.消费消息/*** basicConsume(String queue, boolean autoAck, Consumer callback)* 参数说明:* queue: 队列名称* autoAck: 是否自动确认* callback: 接收到消息后, 执行的逻辑*/DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者接收到消息:" + new String(body));}};channel.basicConsume(Constant.WORK_QUEUE,true,consumer);System.out.println("消费者启动成功");//5.资源释放}
}

观察结果：

3、Publish/Subscribe(发布/订阅)

一个生产者P，多个消费者C1，C2，X代表交换机消息复制多份，每个消费者接收相同的消息生产者发送一条消息，经过交换机转发到多个不同的队列，多个不同的队列就有多个不同的消费者！
适合场景：消息需要被多个消费者同时接收的场景

发布/订阅模式下的交换机为Fanout（扇形）类型

 实现步骤：

  1. 添加依赖
  2. 编写生产者代码
  3. 编写消费者代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写生产者代码：
和前⾯两个的区别是:
需要创建交换机, 并且绑定队列和交换机
//相关配置信息
public class Constant {public static final String HOST = "8.140.60.17";public static final int PORT = 5672;public static final String USER_NAME = "admin";public static final String PASSWORD = "admin";public static final String VIRTUAL_HOST = "xibbei";/*** 发布订阅模式 （Fanout）*/public static final String FANOUT_EXCHANGE ="fanout_exchange" ;public static final String FANOUT_QUEUE1= "fanout_queue1";public static final String FANOUT_QUEUE2= "fanout_queue2";
}*** 扇形（广播）交换机 生产者* 将消息交给所有绑定到交换机的队列(Publish/Subscribe模式)*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel channel = connection.createChannel();//3.声明交换机channel.exchangeDeclare(Constant.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);//4.声明队列channel.queueDeclare(Constant.FANOUT_QUEUE1, true, false, true, null);channel.queueDeclare(Constant.FANOUT_QUEUE2, true, false, true, null);//5.绑定交换机与队列channel.queueBind(Constant.FANOUT_QUEUE1,Constant.FANOUT_EXCHANGE, "");channel.queueBind(Constant.FANOUT_QUEUE2,Constant.FANOUT_EXCHANGE, "");//6.发送消息channel.basicPublish(Constant.FANOUT_EXCHANGE, "", null, "hello world____Fanout".getBytes());System.out.println("发送消息成功");//7.资源释放}}
编写消费者代码：
Routing模式的消费者代码和Publish/Subscribe 代码⼀样, 同样复制出来两份
  • 消费者1:Consumer1
  • 消费者2: Consumer2
修改消费的队列名称就可以
/*** 扇形交换机 一号消费者*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel  channel = connection.createChannel();//3.声明队列channel.queueDeclare(Constant.FANOUT_QUEUE1, true, false, true, null);//4.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1：" + new String(body));}};channel.basicConsume(Constant.FANOUT_QUEUE1,true,consumer);//5.资源释放}
}/*** 扇形交换机 二号消费者*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel channel = connection.createChannel();//3.声明队列channel.queueDeclare(Constant.FANOUT_QUEUE2, true, false, true, null);//4.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2：" + new String(body));}};channel.basicConsume(Constant.FANOUT_QUEUE2,true,consumer);//5.资源释放}
}
观察结果：

4、Routing（路由模式）

路由模式是发布订阅模式的变种，在发布订阅基础上，添加Routing Key

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

路由模式交换机类型为Direct（直出/定向）

 实现步骤：

  1. 添加依赖
  2. 编写生产者代码
  3. 编写消费者代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写生产者代码：
和发布订阅模式的区别是: 交换机类型不同, 绑定队列的BindingKey不同
//配置信息
public class Constant {public static final String HOST = "8.140.60.17";public static final int PORT = 5672;public static final String USER_NAME = "admin";public static final String PASSWORD = "admin";public static final String VIRTUAL_HOST = "xibbei";/*** 路由交换机 （Direct）*/public static final String DIRECT_EXCHANGE = "direct_exchange";public static final String DIRECT_QUEUE1= "direct_queue1";public static final String DIRECT_QUEUE2= "direct_queue2";
}/*** 直连（路由）模式*  把消息交给符合指定routingkey的队列*  生产者*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);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,true,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");//4.发送消息String msg = "hello direct, my routingkey is a....";channel.basicPublish(Constant.DIRECT_EXCHANGE,"a", null, msg.getBytes());String msg_b = "hello direct, my routingkey is b....";channel.basicPublish(Constant.DIRECT_EXCHANGE,"b", null, msg_b.getBytes());String msg_c = "hello direct, my routingkey is c....";channel.basicPublish(Constant.DIRECT_EXCHANGE,"c", null, msg_c.getBytes());System.out.println("消息发送成功");//5.资源释放}}
编写消费者代码：
Routing模式的消费者代码和Publish/Subscribe 代码⼀样, 同样复制出来两份
消费者1:Consumer1
消费者2: Consumer2
修改消费的队列名称就可以

//消费者一号
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel channel = connection.createChannel();//3.声明队列channel.queueDeclare(Constant.DIRECT_QUEUE1,true,false,true,null);//4.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者一号：" + new String(body));}};channel.basicConsume(Constant.DIRECT_QUEUE1, true, consumer);Thread.sleep(1000);System.out.println("消费者一号消费成功");}
}//消费者二号
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.创建通道Channel channel = connection.createChannel();//3.声明队列channel.queueDeclare(Constant.DIRECT_QUEUE2,true,false,true,null);//4.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者二号：" + new String(body));}};channel.basicConsume(Constant.DIRECT_QUEUE2, true, consumer);Thread.sleep(1000);System.out.println("消费者二号消费成功");}
}
观察结果：

5、Topics(通配符模式)

通配符模式是路由模式的升级版，在RoutingKey的基础上，增加了通配符的功能，使之更加灵活

Topics和Routing的基本原理相同，即：生产者将消息发送给交换机，交换机根据routingKey将消息转发给与RoutingKey匹配的队列。 类似于正则表达式

不同点：rooutingKey匹配方式不同 ，Direct是相等匹配 ，topics是通配符匹配

  • * 表⽰⼀个单词
  • # 表⽰多个单词(0-N个）

 实现步骤：

  1. 添加依赖
  2. 编写生产者代码
  3. 编写消费者代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写生产者代码：
//配置信息
public class Constant {public static final String HOST = "8.140.60.17";public static final int PORT = 5672;public static final String USER_NAME = "admin";public static final String PASSWORD = "admin";public static final String VIRTUAL_HOST = "xibbei";/*** 主题交换机 （Topic）*/public static final String TOPIC_EXCHANGE = "topic_exchange";public static final String TOPIC_QUEUE1= "topic_queue1";public static final String TOPIC_QUEUE2= "topic_queue2";public static final String TOPIC_QUEUE3= "topic_queue3";
}/*** 主题交换机（Topic）*  生产者*/
public class Producer {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);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, true, null);channel.queueDeclare(Constant.TOPIC_QUEUE2, true, false, true, null);channel.queueDeclare(Constant.TOPIC_QUEUE3, true, false, true, null);//5.绑定队列channel.queueBind(Constant.TOPIC_QUEUE1, Constant.TOPIC_EXCHANGE, "*.a.*");channel.queueBind(Constant.TOPIC_QUEUE2, Constant.TOPIC_EXCHANGE, "*.a.b");channel.queueBind(Constant.TOPIC_QUEUE3, Constant.TOPIC_EXCHANGE, "c.#");//4.发送消息String msg_a = "Topic模式______ *.*.a";channel.basicPublish(Constant.TOPIC_EXCHANGE, "ac.b.a" ,null, msg_a.getBytes()); //转发消息队列1String msg_b = "Topic模式______ *.a.b";channel.basicPublish(Constant.TOPIC_EXCHANGE, "ab.a.b" ,null, msg_b.getBytes()); //转发消息队列1和消息队列2String msg_c = "Topic模式______ c.*.a";channel.basicPublish(Constant.TOPIC_EXCHANGE, "c.mm" ,null, msg_c.getBytes()); //转发消息队列3System.out.println("发送消息成功");//5.资源释放}}
编写消费者代码：
/*** 主题交换机 一号消费者*/
public class Consumer1 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启通道Channel channel = connection.createChannel();//3.声明交换机 (使用默认交换机)//4.声明队列channel.queueDeclare(Constant.TOPIC_QUEUE1, true, false, true, null);//5.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者1：" + new String(body));}};channel.basicConsume(Constant.TOPIC_QUEUE1,true,consumer);System.out.println("消费者1消费成功");}}
/*** 主题交换机 二号消费者*/
public class Consumer2 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启通道Channel channel = connection.createChannel();//3.声明交换机 (使用默认交换机)//4.声明队列channel.queueDeclare(Constant.TOPIC_QUEUE2, true, false, true, null);//5.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者2：" + new String(body));}};channel.basicConsume(Constant.TOPIC_QUEUE2,true,consumer);System.out.println("消费者2消费成功");}
}/*** 主题交换机 三号消费者*/
public class Consumer3 {public static void main(String[] args) throws IOException, TimeoutException {//1.建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);Connection connection = connectionFactory.newConnection();//2.开启通道Channel channel = connection.createChannel();//3.声明交换机 (使用默认交换机)//4.声明队列channel.queueDeclare(Constant.TOPIC_QUEUE3, true, false, true, null);//5.消费消息DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {System.out.println("消费者3：" + new String(body));}};channel.basicConsume(Constant.TOPIC_QUEUE3,true,consumer);System.out.println("消费者3费成功");}
}
观察结果：

6、RPC（RPC通信）

在RPC通信的过程中, 没有⽣产者和消费者, ⽐较像咱们RPC远程调⽤, ⼤概就是通过两个队列实现了⼀ 个可回调的过程
  1. 客⼾端发送消息到⼀个指定的队列, 并在消息属性中设置replyTo字段, 这个字段指定了⼀个回调队 列, ⽤于接收服务端的响应.
  2. 服务端接收到请求后, 处理请求并发送响应消息到replyTo指定的回调队列
  3. 客⼾端在回调队列上等待响应消息. ⼀旦收到响应，客⼾端会检查消息的correlationId属性，以确保它是所期望的响应

 实现步骤：

  1. 添加依赖
  2. 编写客户端代码
  3. 编写 RPC server代码
引入依赖：
         //引入依赖<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.7.3</version></dependency>
编写客户端代码：
*** rpc 客户端* 1. 发送请求* 2. 接收响应*/
public class RpcClient {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT); //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME);//账号connectionFactory.setPassword(Constant.PASSWORD);  //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();channel.queueDeclare(Constant.RPC_REQUEST_QUEUE, true, false, false, null);channel.queueDeclare(Constant.RPC_RESPONSE_QUEUE, true, false, false, null);//3. 发送请求String msg = "hello rpc...";//设置请求的唯一标识String correlationID = UUID.randomUUID().toString();//设置请求的相关属性AMQP.BasicProperties props = new AMQP.BasicProperties().builder().correlationId(correlationID).replyTo(Constant.RPC_RESPONSE_QUEUE).build();channel.basicPublish("", Constant.RPC_REQUEST_QUEUE, props, msg.getBytes());//4. 接收响应//使用阻塞队列, 来存储响应信息final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String respMsg = new String(body);System.out.println("接收到回调消息: "+ respMsg);if (correlationID.equals(properties.getCorrelationId())){//如果correlationID校验一致response.offer(respMsg);}}};channel.basicConsume(Constant.RPC_RESPONSE_QUEUE, true, consumer);String result = response.take();System.out.println("[RPC Client 响应结果]:"+ result);}
}
编写RPC server代码：
*** RPC server* 1. 接收请求* 2. 发送响应*/
public class RpcServer {public static void main(String[] args) throws IOException, TimeoutException {//1. 建立连接ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT); //需要提前开放端口号connectionFactory.setUsername(Constant.USER_NAME);//账号connectionFactory.setPassword(Constant.PASSWORD);  //密码connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST); //虚拟主机Connection connection = connectionFactory.newConnection();//2. 开启信道Channel channel = connection.createChannel();//3. 接收请求channel.basicQos(1);DefaultConsumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String request = new String(body,"UTF-8");System.out.println("接收到请求:"+ request);String response = "针对request:"+ request +", 响应成功";AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder().correlationId(properties.getCorrelationId()).build();channel.basicPublish("", Constant.RPC_RESPONSE_QUEUE, basicProperties, response.getBytes());channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume(Constant.RPC_REQUEST_QUEUE, false, consumer);}
}

7、Publisher Confirms(发布确认）

发布者确认模式（Publisher Confirmation）是 RabbitMQ 提供的一种确保消息可靠发送RabbitMQ服务器的机制。

用于确保消息被成功发送到交换机（exchange）并被接收到，以及确保消息被正确地路由到队列中。在传统的消息发布过程中，发布者发送消息到交换机后，并不知道消息是否已经被正确地处理。为了解决这个问题，RabbitMQ 提供了发布者确认模式，允许发布者确认消息是否已经被成功接收到。

  1. ⽣产者将Channel设置为confirm模式(通过调⽤channel.confirmSelect()完成)后, 发布的每⼀条消 息都会获得⼀个唯⼀的ID, ⽣产者可以将这些序列号与消息关联起来，以便跟踪消息的状态.
  2. 当消息被RabbitMQ服务器接收并处理后，服务器会异步地向⽣产者发送⼀个确认(ACK)给⽣产者 (包含消息的唯⼀ID)，表明消息已经送达
通过Publisher Confirms模式，⽣产者可以确保消息被RabbitMQ服务器成功接收, 从⽽避免消息丢失的问题.

发布确认模式有三种策略（后续会详细讲解）：

  1. 单独确认
  2. 批量确认
  3. 异步确认

代码实现：

public class PublisherConfirms {private static final Integer MESSAGE_COUNT = 10000;public static Connection createConnection() throws IOException, TimeoutException {ConnectionFactory connectionFactory = new ConnectionFactory();connectionFactory.setHost(Constant.HOST);connectionFactory.setPort(Constant.PORT);connectionFactory.setUsername(Constant.USER_NAME);connectionFactory.setPassword(Constant.PASSWORD);connectionFactory.setVirtualHost(Constant.VIRTUAL_HOST);return connectionFactory.newConnection();}public static void main(String[] args) throws Exception {//1、Publishing Messages Individually//单独确认publishingMessagesIndividually();//2、Publishing Messages in Batches//批量确认publishingMessagesInBatches();//3、Handling Publisher Confirms Asynchronously//异步确认handlingPublishConfirmsAsynchronously();}/*** 异步确认*/private static void handlingPublishConfirmsAsynchronously() throws Exception{try(Connection connection = createConnection()){//1、开启信道Channel channel = connection.createChannel();//2、设置信道confirm模式channel.confirmSelect();//3、声明队列channel.queueDeclare(Constant.PUBLISHER_CONFIRMS_QUEUE3, true, false, false, null);//4、监听confirm//集合中存储的是未确认的消息IDlong start = System.currentTimeMillis();SortedSet<Long> confirmSeqNo = Collections.synchronizedSortedSet(new TreeSet<>());channel.addConfirmListener(new ConfirmListener() {/*** 处理确认消息* @param deliveryTag 确认消息的序列号* @param multiple 是否批量确认* @throws IOException IO异常*/@Overridepublic void handleAck(long deliveryTag, boolean multiple) throws IOException {if (multiple) {// 批量确认：移除所有小于等于当前deliveryTag的消息记录confirmSeqNo.headSet(deliveryTag + 1).clear();} else {// 单条确认：仅移除当前消息记录confirmSeqNo.remove(deliveryTag);}}/*** 处理未确认消息（即发送失败的消息）* @param deliveryTag 未确认消息的序列号* @param multiple 是否批量未确认* @throws IOException IO异常*/@Overridepublic void handleNack(long deliveryTag, boolean multiple) throws IOException {if (multiple) {// 批量未确认：移除所有小于等于当前deliveryTag的消息记录confirmSeqNo.headSet(deliveryTag + 1).clear();} else {// 单条未确认：仅移除当前消息记录confirmSeqNo.remove(deliveryTag);}// 业务需要根据实际场景进行处理，比如重发，此处代码省略}});//5、发送消息for (int i = 0; i < MESSAGE_COUNT; i++) {String msg = "hello publisher confirms"+i;long seqNo = channel.getNextPublishSeqNo();channel.basicPublish("",Constant.PUBLISHER_CONFIRMS_QUEUE3, null, msg.getBytes());confirmSeqNo.add(seqNo);}while (!confirmSeqNo.isEmpty()){Thread.sleep(10);}long end = System.currentTimeMillis();System.out.printf("异步确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end-start);}}/*** 批量确认*/private static void publishingMessagesInBatches() throws Exception{try(Connection connection = createConnection()){//1、开启信道Channel channel = connection.createChannel();//2、设置信道confirm模式channel.confirmSelect();//3、声明队列channel.queueDeclare(Constant.PUBLISHER_CONFIRMS_QUEUE2, true, false, false, null);//4、发送消息,并等待确认long start = System.currentTimeMillis();int batchSize = 100;int outstandingMessageCount = 0;for(int i = 0;i<MESSAGE_COUNT;i++){String msg = "hello publisher confirms"+i;channel.basicPublish("",Constant.PUBLISHER_CONFIRMS_QUEUE2, null, msg.getBytes());outstandingMessageCount++;if(outstandingMessageCount == batchSize){channel.waitForConfirmsOrDie(500);outstandingMessageCount = 0;}}if(outstandingMessageCount >0){channel.waitForConfirmsOrDie(500);}long end = System.currentTimeMillis();System.out.printf("批量确认策略, 批量大小: %d, 消息条数: %d, 耗时: %d ms \n",batchSize, MESSAGE_COUNT, end-start);}}/*** 单独确认*/private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {try(Connection connection = createConnection()){//1、开启信道Channel channel = connection.createChannel();//2、设置信道confirm模式channel.confirmSelect();//3、声明队列channel.queueDeclare(Constant.PUBLISHER_CONFIRMS_QUEUE1, true, false, false, null);//4、发送消息,并等待确认long start = System.currentTimeMillis();for(int i = 0;i<MESSAGE_COUNT;i++){String msg = "hello publisher confirms"+i;channel.basicPublish("",Constant.PUBLISHER_CONFIRMS_QUEUE1, null, msg.getBytes());//等待确认channel.waitForConfirmsOrDie(500);}long end = System.currentTimeMillis();System.out.printf("单独确认策略, 消息条数: %d, 耗时: %d ms \n",MESSAGE_COUNT, end-start);}}
}

http://www.dtcms.com/a/321212.html

相关文章：

  • springmvc4+hibernate4整合框架的搭建
  • 库函数版独立按键用位运算方式实现（STC8）
  • Array Description(Dynamic programming)
  • 【网络编程】IO多路转接——epoll
  • Java文件读写（IO、NIO）
  • 第39周——训练自己的数据集
  • 汇编语言和高级语言的差异
  • BGP综合实验练习作业
  • Fabarta个人专属智能体：三维搜索链+动态大纲重构教材开发范式
  • Omron（欧姆龙）SysmacStudio软件下载，定期更新(最新更新到1.63升级包)
  • npm run 常见脚本
  • BGP协议笔记
  • 【新启航】航空飞机起落架深孔型腔的内轮廓测量方法探究 - 激光频率梳 3D 轮廓检测
  • 2025华数杯数学建模A题【 多孔膜光反射性能的优化与控制】原创论文讲解（含完整python代码）
  • 避免“卡脖子”！如何减少内存I/O延迟对程序的影响？
  • 机器学习——支持向量机（SVM）实战案例
  • 操作系统-实验-进程
  • 机器学习之支持向量机(原理)
  • svm的一些应用
  • 怎么查看Linux I2C总线挂载了那些设备？
  • springboot整合rabbitMQ的示例
  • Elasticsearch：在向量搜索中使用 Direct IO
  • 解码华为云安全“铁三角”：用“分层防御”化解安全挑战
  • 微软披露Exchange Server漏洞：攻击者可静默获取混合部署环境云访问权限
  • 企业AI的双层技术栈架构：融合社区创新与企业级管控的设计蓝图
  • Git 使用场景笔记
  • DuoPlus支持导入文件批量配置云手机参数，还优化了批量操作和搜索功能！
  • 数据结构--哈希表
  • QAGenerationChain从知识库生成大模型应用测试的问题对
  • LeetCode算法日记 - Day 5: 长度最小的子数组、无重复字符的最长子串