小众网站论文百度提交入口网址在哪
目录
一. 作用：
二. RabbitMQ的5中队列模式：
1. 简单模式
2. Work模式
3. 发布/订阅模式
4. 路由模式
5. 主题模式
三. 消息持久化：
消息过期时间
ACK应答
四. 同步接收和异步接收：
应用场景
五. 基本使用 ：
引入依赖库：
配置文件RabbitMQConfig：
创建消息任务类：
解析：
一. 作用：
RabbitMQ主要用于消息队列的实现。
二. RabbitMQ的5中队列模式：
1. 简单模式
一个生产者（发送方）对应一个消费者（接收方）
2. Work模式
一个生产者对应多个消费者，但是只能有一个消费者获得消息（排他）
3. 发布/订阅模式
一个消费者将消息首先发送到fanout交换器，交换器绑定到多个队列，然后与之对应的所有消费者都能接收到消息（不排他）
4. 路由模式
生产者将消息发送到direct交换器，交换器按照关键字（Key），把消息路由到某个队列
5. 主题模式
生产者将消息发送到Topic交换器，交换器按照复杂的规则，把消息路由到某个队列
三. 消息持久化：
消息的可靠性是RabbitMQ的一大特色，那么RabbitMQ是如何保证消息可靠性的呢？答案就是消息持久化。持久化可以防止在异常情况下丢失数据。除了消息持久化之外，甚至交换器和队列都能持久化。也就是说rabbitmq的消息会被存储在磁盘上，只有当消费收到消息，rabbitmq确认消费者收到消息（Acknowledgments--简称ACK）后才会将消息从队列中删除。
-
消息过期时间
如果消费者一直不接收消息，消息会一直保存在消息队列当中，短期内可能不会有什么影响，但是如果经过长时间的积累后，消息会变得很多很多 ，浪费大量的资源，内存。
为了应对这种情况，就可以对rabbitmq设置消息的过期时间，在规定时间内消息没有被接收，就会删除掉该消息。
-
ACK应答
消费者接收到消息后，为了让RabbitMQ 知道，就需要返回一个ACK应答，告诉RabbitMQ消费者已经收到了消息，如果收到消息后我们需要删除该消息，只需要在ACK应答中加上deliveryTag标志位。
四. 同步接收和异步接收：
同步接收：指消费者调用方法时，会阻塞来等待消息，直到消息被成功消费或者队列为空。（没有消息等待消息再接着处理）。
异步接收： 指消费者不会在接收消息时阻塞，而是通过回调函数处理消息。消费者在等待消息的同时不会停下，可以处理其他任务。（当有消息时才来处理消息）。
-
应用场景
同步接收 ：当消息的处理顺序对业务逻辑非常重要，就使用同步接收，消费者一次只处理一个消息，确保了每条消息的处理顺序。
异步接收：当处理消息的时间比较长，或者系统的并发量大时，采用异步接收会更好。
RabbitMQ还有一个杀手锏——同时使用异步收发和同步收发。
五. 基本使用 ：
引入依赖库：
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>5.9.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
配置文件RabbitMQConfig：
import com.rabbitmq.client.ConnectionFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitMQConfig {@Value("${rabbitmq.factoryHost}")private String host;@Beanpublic ConnectionFactory connectionFactory() {ConnectionFactory factory = new ConnectionFactory();factory.setHost(host);factory.setPort(5672);return factory;}
}host配置，我将rabbitMQ放在虚拟机上的，所有ip是虚拟机的地址：
创建消息任务类：
@Slf4j
@Component
public class MessageTask {@Autowiredprivate ConnectionFactory factory;@Autowiredprivate MessageService messageService;/** 同步发送消息* */public void send(String topic, MessageEntity entity) {//向MongoDB保存消息数据，返回消息IDString id = messageService.insertMessage(entity);//向RabbitMQ发送消息try(Connection connection = factory.newConnection();Channel channel = connection.createChannel()){//连接到某个topicchannel.queueDeclare(topic, true, false, false, null);HashMap header = new HashMap();header.put("messageId",id);//创建AMQP协议参与对象，添加附加属性AMQP.BasicProperties properties = new AMQP.BasicProperties().builder().headers(header).build();channel.basicPublish("",topic,properties,entity.getMsg().getBytes());log.debug("消息发送成功");} catch (Exception e){log.error(e.getMessage());throw new EmosException("向MQ发送消息失败");}}/** 异步发送消息* */@Async("AsyncTaskExecutor")public void sendAsync(String topic, MessageEntity entity) {send(topic, entity);}/** 同步接收消息* */public int receive(String topic) {int i = 0;try (//接收消息数据Connection connection = factory.newConnection();Channel channel = connection.createChannel()) {// 从队列中获取消息，不自动确认channel.queueDeclare(topic, true, false, false, null);//Topic中有多少条数据未知，所以使用死循环接收数据，直到接收不到消息，退出死循环while (true) {//创建响应接收数据，禁止自动发送Ack应答GetResponse response = channel.basicGet(topic, false);if (response != null) {AMQP.BasicProperties properties = response.getProps();Map<String, Object> header = properties.getHeaders(); //获取附加属性对象String messageId = header.get("messageId").toString();byte[] body = response.getBody();//获取消息正文String message = new String(body);log.debug("从RabbitMQ接收的消息：" + message);MessageRefEntity entity = new MessageRefEntity();entity.setMessageId(messageId);entity.setReceiverId(Integer.parseInt(topic));entity.setReadFlag(false);entity.setLastFlag(true);messageService.insertRef(entity); //把消息存储在MongoDB中//数据保存到MongoDB后，才发送Ack应答，让Topic删除这条消息long deliveryTag = response.getEnvelope().getDeliveryTag();channel.basicAck(deliveryTag, false);i++;} else {break; //接收不到消息，则退出死循环}}} catch (Exception e) {log.error("执行异常", e);}return i;}/** 异步接收消息* */@Asyncpublic int receiveAsync(String topic) {return receive(topic);}/** 同步删除消息* */public void deleteQueue(String topic) {try(//接收消息数据Connection connection = factory.newConnection();Channel channel = connection.createChannel()){channel.queueDelete(topic);log.debug("成功删除消息队列:"+topic);} catch (Exception e){log.error("删除消息队列失败:",e);throw new EmosException("删除消息队列失败");}}/** 异步删除消息* */@Asyncpublic void deleteAsync(String topic) {deleteQueue(topic);}
}解析：
channel.queueDeclare(String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments);
- queueName:队列的名称，用于标识消息的存储位置。
- durable：
true，表示队列是持久化的。
false，表示队列是非持久化的。
- exclusive：
true：队列仅供当前连接使用，连接断开时队列会自动删除。
false：队列可供多个连接共享。
-
autoDelete：
true：当队列不再被任何消费者订阅时，队列会自动删除。
false：队列即使没有消费者订阅也会一直存在，直到手动删除。 - arguments：额外的参数，null表示没有额外参数
Map<String, Object> arguments = new HashMap<>();
arguments.put("x-message-ttl", 60000); // 设置队列中消息的过期时间为 60 秒（60000 毫秒）channel.queueDeclare("myQueue", true, false, false, arguments);