RabbitMQ个人理解与基本使用

目录

一. 作用：

二. 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);