广州网站开发 找亦客公司优质西安膜结构
目录
简单模式
定义
工作流程
应用场景
优点
缺点
小结
使用案例
引入依赖
编写生产者代码
创建连接
创建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();}
}运行程序
原因分析及解决办法
明明队列中有10条消息，而且队列中的消息条数也已经由10条变成了0条，这是为什么呢？
----------是因为我们在接收消息时设置了自动确认，但是接收到的消息还没有全都打印出来之前，连接已经关闭了，程序已经结束了。
----------解决方法，连接关闭之前，先强制等待一段时间。
运行结果
此时我们可以看到，运行结果符合预期。