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

详解RabbitMQ工作模式之简单模式

news 2025/7/2 1:49:53

简单模式

定义

工作流程

应用场景

优点

缺点

小结

使用案例

引入依赖

编写生产者代码

创建连接

创建Channel

声明一个队列

发送消息

释放资源

整体代码

管理界面观察队列消息

编写消费者代码

创建连接+创建Channel+声明队列

消费消息

释放资源

整体代码

运行程序

原因分析及解决办法

简单模式

定义

简单模式（Queue模式）是RabbitMQ中最基础的消息传递模式。在这种模式下，生产者将消息发送到交换机，交换机根据消息属性将其发送到队列，消费者则监听并绑定到该队列，从而接收并处理消息。简单模式强调的是一个队列只被一个消费者监听和消费。

P: ⽣产者, 也就是要发送消息的程序
C: 消费者,消息的接收者
Queue: 消息队列, 图中⻩⾊背景部分. 类似⼀个邮箱, 可以缓存消息; ⽣产者向其中投递消息, 消费者从其中取出消息

工作流程

1.生产者生成一条消息，并将其发送到交换机。
2.交换机根据消息属性（在简单模式下可能是固定的路由键或不需要路由键）将消息发送到指定的队列。
3.消费者监听并绑定到该队列，当队列中有消息时，消费者将其取出并处理。
4.消息被消费者处理后，会自动从队列中删除（除非设置了消息持久化或手动确认机制）。

应用场景

简单模式适用于那些需要一对一消息传递的场景，例如：

1.手机短信发送：一个生产者（如短信服务）将短信内容发送到交换机，交换机将其发送到指定的队列，消费者（如短信网关）从队列中接收短信并发送到用户的手机上。
2.邮件单发：一个生产者（如邮件服务）将邮件内容发送到交换机，交换机将其发送到指定的队列，消费者（如邮件发送器）从队列中接收邮件并发送到指定的邮箱。

优点

1.结构简单，易于理解和实现。
2.消息传递可靠，确保消息按照到达的顺序被处理。

缺点

1.一个队列只能被一个消费者监听和消费，可能无法充分利用系统资源。
2.在高并发场景下，单个消费者可能无法及时处理所有消息，导致消息堆积。

小结

RabbitMQ的简单模式是一种基础且可靠的消息传递模式，适用于那些需要一对一消息传递的场景。它通过生产者、交换机、队列和消费者之间的协作，实现了消息的可靠传递和处理。然而，在高并发或需要充分利用系统资源的场景下，可能需要考虑使用更复杂的消息传递模式（如工作队列模式、发布订阅模式等）来满足需求。

使用案例

引入依赖

<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->
<dependency><groupId>com.rabbitmq</groupId><artifactId>amqp-client</artifactId><version>5.21.0</version>
</dependency>

编写生产者代码

创建连接

        ConnectionFactory factory = new ConnectionFactory();factory.setHost("47.98.109.138");factory.setPort(5672);factory.setUsername("study");factory.setPassword("study");factory.setVirtualHost("aaa");Connection connection = factory.newConnection();

创建Channel

⽣产者和消费者创建的channel并不是同⼀个

Channel channel = connection.createChannel();

声明一个队列

        /*** queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete,*                                  Map<String, Object> arguments)*  参数说明:*  queue: 队列名称*  durable: 可持久化*  exclusive: 是否独占*  autoDelete: 是否自动删除*  arguments: 参数*/channel.queueDeclare("hello", true, false, false, null);

如果有⼀个名为 "hello" 的队列, ⽣产者可以直接发送消息到 "hello" 队列, ⽽消费者可以从
"hello" 队列中接收消息, ⽽不需要关⼼交换机的存在. 这种模式⾮常适合简单的应⽤场景，其中⽣产者和消费者之间的通信是⼀对⼀的。

发送消息

        /*** basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)* 参数说明:* exchange: 交换机名称* routingKey: 内置交换机, routingkey和队列名称保持一致* props: 属性配置* body: 消息*/for (int i = 0; i < 10; i++) {String msg = "hello rabbitmq~"+i;channel.basicPublish("","hello", null, msg.getBytes());}

释放资源

    channel.close();connection.close();

整体代码

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ProducerDemo {public static void main(String[] args) throws IOException, TimeoutException {ConnectionFactory factory = new ConnectionFactory();factory.setHost("47.98.109.138");factory.setPort(5672);factory.setUsername("study");factory.setPassword("study");factory.setVirtualHost("aaa");Connection connection = factory.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, false, 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 rabbitmq~"+i;channel.basicPublish("","hello", null, msg.getBytes());}System.out.println("消息发送成功~");//6. 资源释放channel.close();connection.close();}
}

管理界面观察队列消息

可以看到此时名为“hello”的队列中有10条消息。

编写消费者代码

创建连接+创建Channel+声明队列

        ConnectionFactory factory = new ConnectionFactory();factory.setHost("47.98.109.138");factory.setPort(5672);factory.setUsername("study");factory.setPassword("study");factory.setVirtualHost("aaa");Connection connection = factory.newConnection();Channel channel = connection.createChannel();channel.queueDeclare("hello",true, false, false, null);

消费消息

        /*** 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);

释放资源

    channel.close();connection.close();

整体代码

import com.rabbitmq.client.*;import java.io.IOException;
import java.util.concurrent.TimeoutException;public class ConsumerDemo {public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {//1. 创建连接ConnectionFactory factory = new ConnectionFactory();factory.setHost("47.98.109.138");factory.setPort(5672);factory.setUsername("study");factory.setPassword("study");factory.setVirtualHost("aaa");Connection connection = factory.newConnection();//2. 创建ChannelChannel channel = connection.createChannel();//3. 声明队列(可以省略)channel.queueDeclare("hello",true, false, false, 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);//5. 释放资源channel.close();connection.close();}
}