目录
一. 作用：
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;
@Bean
public ConnectionFactory connectionFactory() {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost(host);
factory.setPort(5672);
return factory;
}
}host配置，我将rabbitMQ放在虚拟机上的，所有ip是虚拟机的地址：
创建消息任务类：
@Slf4j
@Component
public class MessageTask {
@Autowired
private ConnectionFactory factory;
@Autowired
private MessageService messageService;
/*
* 同步发送消息
* */
public void send(String topic, MessageEntity entity) {
//向MongoDB保存消息数据，返回消息ID
String id = messageService.insertMessage(entity);
//向RabbitMQ发送消息
try(Connection connection = factory.newConnection();
Channel channel = connection.createChannel()){
//连接到某个topic
channel.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;
}
/*
* 异步接收消息
* */
@Async
public 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("删除消息队列失败");
}
}
/*
* 异步删除消息
* */
@Async
public 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);