SpringBoot2.X 整合RabbitMQ
spring-boot-starter-amqp
官方文档地址
https://spring.io/projects/spring-amqp
image.png
官方文档地址
https://docs.spring.io/spring-amqp/docs/current/reference/html/
image.png
RabbitMQ简介
RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。
RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。 RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。
AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。
RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。
三个概念
有三个概念： 发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和 队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。
官网截图
官网截图.png
- 左侧 P 代表 生产者，也就是往 RabbitMQ 发消息的程序。
- 中间即是 RabbitMQ，其中包括了 交换机 和 队列。
- 右侧 C 代表 消费者，也就是往 RabbitMQ 拿消息的程序。
重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。
交换机(Exchange)
交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用ack模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout：
Direct：direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
Topic：按规则转发消息（最灵活）
Headers：设置 header attribute 参数类型的交换机
Fanout：转发消息到所有绑定队列
Direct Exchange 是 RabbitMQ 默认的交换机模式，也是最简单的模式，根据key全文匹配去寻找队列。
Topic Exchange 转发消息主要是根据通配符。* 在这种交换机下，队列和交换机的绑定会定义一种路由模式，那么，通配符就要在这种路由模式和路由键之间匹配后交换机才能转发消息。
两个通配符:
-
*表示一个词. -
#表示零个或多个词.
Headers Exchange是一个自定义匹配规则的类型. 在队列与交换器绑定时, 会设定一组键值对规则, 消息中也包括一组键值对( headers 属性), 当这些键值对有一对, 或全部匹配时, 消息被投送到对应队列.
Fanout ExchangeFanout Exchange 消息广播的模式，不管路由键或者是路由模式，会把消息发给绑定给它的全部队列，如果配置了 routing_key 会被忽略。
Queue 队列
是RabbitMQ的内部对象，用于存储消息。
Binding 绑定
RabbitMQ通过绑定将交换机与队列关联起来，在绑定的时候一般会指定一个绑定建（BindingKey），这样RabbitMQ就会将消息路由到队列。
RabbitMQ持久化机制
队列持久化，消息持久化，交换机持久化
队列持久化
在定义队列的时durable参数来实现，当durable为true时，队列才会持久化。
image.png
/**
* Construct a new queue, given a name, durability, exclusive and auto-delete flags.
* @param name 队列名称.
* @param true如果我们声明一个持久队列（该队列将在服务器重新启动后继续存在）
* @param true如果要声明独占队列,（该队列将仅由声明者的连接使用）
* @param true如果服务器应在队列不再使用时将其删除，则自动删除
*/
public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete) {
this(name, durable, exclusive, autoDelete, null);
}
消息持久化
消息持久化通过消息的属性deliveryMode来设置是否持久化.
Message message = MessageBuilder.withBody(JsonUtil.objToStr(obj).getBytes()).build();
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);//消息持久化
message.getMessageProperties().setContentType(MessageProperties.CONTENT_TYPE_JSON);
image.png
交换机持久化
通过属性durable设置，默认为true，即交换机持久化。
image.png
/**
* 构造一个新交换机,名称, 持久化flag, 自动删除 flag.
* @param 交换机名称
* @param true 声明交换机持久化(交换机将服务器重新启动后继续运行)
* @param true 交换机在服务器不在使用则自动删除
*/
public AbstractExchange(String name, boolean durable, boolean autoDelete) {
this(name, durable, autoDelete, null);
}
SpringBoot 集成RabbitMQ示例
1 . pom导入依赖
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2 . application配置
spring:
# RabbitMQ
rabbitmq:
#rabbitmq的连接地址
host: localhost
#rabbitmq的连接端口号
port: 5672
#rabbitmq的虚拟host
virtual-host: /test
username: guest
password: guest
#如果对异步消息需要回调必须设置为true
publisher-confirms: true
3. 队列配置
@Getter
public enum QueueEnum {
/**
* 消息通知队列
*/
QUEUE_DEMO_CANCEL("test.demo.direct",
"test.order.cancel",
"test.order.cancel"),
/**
* 消息通知ttl队列
*/
QUEUE_TTL_DEMO_CANCEL("test.demo.direct.ttl",
"test.demo.cancel.ttl",
"test.demo.cancel.ttl");
}
/**
* 消息队列配置
*
*/
@Configuration
public class RabbitMqConfig {
/**
* 订单消息实际消费队列所绑定的交换机
*
* @return
*/
@Bean
DirectExchange orderDirect(){
return (DirectExchange) ExchangeBuilder
.directExchange(QueueEnum.QUEUE_DEMO_CANCEL.getExchange())
.durable(true)
.build();
}
/**
* 订单延迟队列队列所绑定的交换机
* @return
*/
@Bean
DirectExchange orderTtlDirect(){
return (DirectExchange) ExchangeBuilder
.directExchange(QueueEnum.QUEUE_TTL_DEMO_CANCEL.getExchange())
.durable(true)
.build();
}
/**
* 订单实际消费队列
* @return
*/
@Bean
public Queue orderQueue(){
return new Queue(QueueEnum.QUEUE_DEMO_CANCEL.getName());
}
/**
* 死信队列 场景：消费失败、消息过期、队列超载的消息
* 订单延迟队列（死信队列）
* 死信队列
* x-dead-letter-exchange
* x-dead-letter-routing-key
* @return
*/
@Bean
public Queue orderTtlQueue(){
return QueueBuilder
.durable(QueueEnum.QUEUE_TTL_DEMO_CANCEL.getName())
.withArgument("x-dead-letter-exchange",QueueEnum.QUEUE_DEMO_CANCEL.getExchange())//到期后转发的交换机
.withArgument("x-dead-letter-routing-key",QueueEnum.QUEUE_DEMO_CANCEL.getRouteKey())//到期后转发的路由键
.build();
}
/**
* 将 订单队列 绑定 到交换机
*/
@Bean
Binding orderBinding(DirectExchange orderDirect,Queue orderQueue){
return BindingBuilder
.bind(orderQueue)
.to(orderDirect)
.with(QueueEnum.QUEUE_DEMO_CANCEL.getRouteKey());
}
/**
* 将订单延迟队列绑定到交换机
*/
@Bean
Binding orderTtlBinding(DirectExchange orderTtlDirect,Queue orderTtlQueue){
return BindingBuilder
.bind(orderTtlQueue)
.to(orderTtlDirect)
.with(QueueEnum.QUEUE_TTL_DEMO_CANCEL.getRouteKey());
}
}
发送消息
/**
* 取消消息的发出者
*/
@Component
public class CancelemoSender {
private static Logger LOGGER = LoggerFactory.getLogger(CancelemoSender .class);
@Autowired
private AmqpTemplate amqpTemplate;
/**
*
* @param orderId 订单号
* @param delayTimes 延迟时间
*/
public void sendMessage(Long orderId,final long delayTimes){
//给延迟队列发送消息
amqpTemplate.convertAndSend(QueueEnum.QUEUE_TTL_DEMO_CANCEL.getExchange(),
QueueEnum.QUEUE_TTL_DEMO_CANCEL.getRouteKey(),
orderId,
new MessagePostProcessor() {//在发送消息之前，携带处理消息
@Override
public Message postProcessMessage(Message message) throws AmqpException {
//给消息设置延迟毫秒值
message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
return message;
}
});
LOGGER.info("send delay message orderId:{}",orderId);
}
}
收到消息
/**
* 取消订单消息的处理者
*/
@Component
@RabbitListener(queues = "test.demo.cancel")
public class CancelDemoReceiver {
private static Logger LOGGER = LoggerFactory.getLogger(CancelDemoReceiver .class);
@RabbitHandler
public void handle(Long orderId){
LOGGER.info("receive delay message orderId:{}",orderId);
}
}
消息发送确认机制
/**
* TopicExchange 发送消息
*/
@Slf4j
@Component //需要注入到Spring容器中
public class TopicSender {
@Autowired
private RabbitTemplate rabbitTemplate;//注入一个RabbitMQ的模板对象，操作消息队列的对象
/**
* exchange 交换机
*/
public void sendExchange1(String text){
//向交换机exchange里的路由秘钥routingKey发送消息,通过路由秘钥发送到对应的消息队列
this.rabbitTemplate.convertAndSend(TopicRabbitConfig.EXCHANGE,
TopicRabbitConfig.routingKey,text);
}
/**
* exchange 交换机
*/
public void sendExchange2(String text){
//向交换机exchange里的路由秘钥routingKey发送消息,通过路由秘钥发送到对应的消息队列
this.rabbitTemplate.convertAndSend(TopicRabbitConfig.EXCHANGE,
TopicRabbitConfig.routingKey_s,text);
}
/**
* 确认消息 是否发送到exchange交换机
* @param text
*/
public void sendExchange3(String text){
//向交换机exchange里的路由秘钥routingKey发送消息,通过路由秘钥发送到对应的消息队列
// this.rabbitTemplate.convertAndSend(TopicRabbitConfig.EXCHANGE,
// TopicRabbitConfig.routingKey,text);
// properties=BasicProperties(
// delivery_mode=2 // make message persistent
//确保消息发送失败后可以重新返回到队列中
//yml文件 spring.rabbitmq.publisher-returns=true
this.rabbitTemplate.setMandatory(true);
//使用confirm确认机制
// 消息是否成功发送到Exchange
this.rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if (ack){
log.info("消息成功发送到Exchange");
}else {
log.info("消息发送到Exchange失败: cause: {}", correlationData, cause);
}
}
});
/**
* #开启returnedMessage回调 Exchange -> Queue
* spring.rabbitmq.publisher-returns=true
* #触发returnedMessage回调必须设置mandatory=true,
* # 否则Exchange没有找到Queue就会丢弃掉消息, 而不会触发回调
* spring.rabbitmq.template.mandatory=true
*/
// 消息是否从Exchange路由到Queue, 注意: 这是一个失败回调, 只有消息从Exchange路由到Queue失败才会回调这个方法
this.rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("消息从Exchange路由到Queue失败: exchange: {}, route: {}, replyCode: {}, replyText: {}, message: {}",
exchange, routingKey, replyCode, replyText, message);
}
});
Message message = MessageHelper.objToMsg(text);//消息持久化
this.rabbitTemplate.convertAndSend(TopicRabbitConfig.EXCHANGE,
TopicRabbitConfig.routingKey_s,message,
new CorrelationData(UUID.randomUUID().toString()));
}
}
@Configuration
@RabbitListener(queues = TopicRabbitConfig.message)
public class Topic1Receiver {
// @RabbitHandler
// public void process(String text){
// System.out.println("Topic 1 Receiver:"+text);
// }
/**
* # 设置手动确认(ack) Queue -> C ,消息是否被消费;
* # NONE：自动确认;AUTO：根据情况确认;MANUAL：手动确认
* spring.rabbitmq.listener.simple.acknowledge-mode=manual
*
* @param message
* @param channel
* @param tag
*/
@RabbitHandler
public void process2(String msg,Message message, Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
try {
// log.info("Topic 1 Receiver process 1 收到消息：msg {}", msg.toString());
log.info("Topic 1 Receiver process 1 收到消息：message {}", message.getBody());
///做一些操作：数据库操作等
}catch (Exception e){
channel.basicNack(tag,false,true);
}finally {
channel.basicAck(tag,false);
}
// 消费消息并手动确认:
}
/**
* # 设置手动确认(ack) Queue -> C ,消息是否被消费;
* # NONE：自动确认;AUTO：根据情况确认;MANUAL：手动确认
* spring.rabbitmq.listener.simple.acknowledge-mode=manual
*
* @param message
* @param channel
* @param tag
*/
@RabbitListener(queues = TopicRabbitConfig.messages)
public void process3(String msg,Message message, Channel channel,
@Header(AmqpHeaders.DELIVERY_TAG) long tag) {
try {
log.info("Topic 2 Receiver process 3 收到消息：msg {}", msg);
// log.info("Topic 2 Receiver process 3 收到消息：message {}", message.getBody().toString());
///做一些操作：数据库操作等
}catch (Exception e){
//消息即将再次返回队列处理
try {
channel.basicNack(tag,false,true);
} catch (IOException ex) {
ex.printStackTrace();
log.info("Topic 2 Receiver process 3 basicNack IOException {}", ex.getMessage());
}
}finally {
//
try {
channel.basicAck(tag,false);
} catch (IOException e) {
e.printStackTrace();
log.info("Topic 2 Receiver process 3 basicAck IOException {}", e.getMessage());
}
}
// 消费消息并手动确认:
}
}
