消息队列应用场景

消息中间件概述

RabbitMQ 概念

一个客户端只与消息中间件建立一条链接（长连接），一条链接里面有多个channel（虚拟链接，复用一条TCP连接）

虚拟主机是rabbitmq为了隔离不同类型客户端出现的概念。我们在开发过程中可以一个客户端一个虚拟主机。即一个rabbitmq里面可以有多个虚拟主机

虚拟主机：多个交换机构成虚拟主机，主要目的是隔离客户端的消息。比如当前消息队列的生产者有java端和php端，为了隔离开我们可以划分出两个虚拟主机。虚拟主机互相隔离，一台虚拟主机出现问题，不会影响到别的虚拟主机

虚拟主机是按照路径来划分的。

流程：无论是生产者往mq发送消息，还是消费者接受mq的消息，都会建立一条连接，所有的数据都会在连接里面开辟信道来进行收发。收发的消息分成两部分，消息头和消息体。消息头就是消息的属性信息，消息体就是消息的真正内容，消息里面还需要定义发给那个交换机，那个虚拟主机，消息头里面最重要就是route-key。发消息时，消息会来到rabbitmq里面指定的虚拟主机中，然后到达指定的交换机。交换机通过消息里面的route-key和绑定的队列的路由键，决定将消息发送给与交换机绑定的哪些队列中。如果队列放入的消息，监听队列的消费者就会得到队列里面的数据。一个客户端建立一条长连接的好处有，mq能实时的知道那个消费端断线了，从而及时的将队列里面的数据保存起来（停止出队），防止消息丢失。

docker 安装rabbitmq

docker run -d \
-p 5671:5671 \
-p 5672:5672 \
-p 4369:4369 \
-p 25672:25672 \
-p 15671:15671 \
-p 15672:15672 \
--restart=always --name rabbitmq rabbitmq:3-management默认账户密码是guest
4369， 25672 （Erlang发现&集群端口）
5672， 5671 （AMQP端口）
15672 （web管理后台端口）
61613， 61614 （STOMP协议端口）
1883， 8883 （MQTT协议端口）
https://www.rabbitmq.com/networking.html

RabbitMQ的运行机制

一个交换机可以绑定多个队列，一个队列可以绑定多个交换机。交换机决定要按照什么绑定关系路由给哪个消息队列。

交换机的类型：direct、fanout（扇出）、topic、headers。direct和headers 是点对点的实现、fanout和topic是发布订阅模式的实现。而header性能低下，现在都只用direct、fanout、topic这三种不同类型的交换机。

发消息是将消息发送给虚拟主机里面的交换机，消费消息是监听队列。

direct：直接类型交换机。按照路由键精确匹配到一个队列。

fanout：广播类型交换机。消息会发送给交换机下面绑定的所有队列，不管路由键。

topic：发布订阅模式交换机。可根据路由键进行模糊匹配。#匹配0个或多个单词，* 匹配一个单词。

注意：路由键是有单词组成的，单词之间使用 . 分割，这一点是理解topic模糊匹配的关键。

测试一下三种不同类型的exchange

导出的配置（是json文件，xxx.json 导入即可）

{"rabbit_version":"3.8.5","rabbitmq_version":"3.8.5","product_name":"RabbitMQ","product_version":"3.8.5","users":[{"name":"guest","password_hash":"VLq+EgZQzd1z32LOZ83Onxc/MFCJcoMPf4jynGCdC9Aqvc6B","hashing_algorithm":"rabbit_password_hashing_sha256","tags":"administrator"}],"vhosts":[{"name":"/"}],"permissions":[{"user":"guest","vhost":"/","configure":".*","write":".*","read":".*"}],"topic_permissions":[],"parameters":[],"global_parameters":[{"name":"cluster_name","value":"rabbit@7072bb66c978"},{"name":"internal_cluster_id","value":"rabbitmq-cluster-id-9wl1WHjPlMGESsZsgSTH_g"}],"policies":[],"queues":[{"name":"atguigu.news","vhost":"/","durable":true,"auto_delete":false,"arguments":{"x-queue-type":"classic"}},{"name":"atguigu","vhost":"/","durable":true,"auto_delete":false,"arguments":{"x-queue-type":"classic"}},{"name":"gulixueyuan.news","vhost":"/","durable":true,"auto_delete":false,"arguments":{"x-queue-type":"classic"}},{"name":"atguigu.emps","vhost":"/","durable":true,"auto_delete":false,"arguments":{"x-queue-type":"classic"}}],"exchanges":[{"name":"exchange.fanout","vhost":"/","type":"fanout","durable":true,"auto_delete":false,"internal":false,"arguments":{}},{"name":"exchange.topic","vhost":"/","type":"topic","durable":true,"auto_delete":false,"internal":false,"arguments":{}},{"name":"my.exchange.direct","vhost":"/","type":"direct","durable":true,"auto_delete":false,"internal":false,"arguments":{}},{"name":"exchange.direct","vhost":"/","type":"direct","durable":true,"auto_delete":false,"internal":false,"arguments":{}}],"bindings":[{"source":"exchange.direct","vhost":"/","destination":"atguigu","destination_type":"queue","routing_key":"atguigu","arguments":{}},{"source":"exchange.direct","vhost":"/","destination":"atguigu.emps","destination_type":"queue","routing_key":"atguigu.emps","arguments":{}},{"source":"exchange.direct","vhost":"/","destination":"atguigu.news","destination_type":"queue","routing_key":"atguigu.news","arguments":{}},{"source":"exchange.direct","vhost":"/","destination":"gulixueyuan.news","destination_type":"queue","routing_key":"gulixueyuan.news","arguments":{}},{"source":"exchange.fanout","vhost":"/","destination":"atguigu","destination_type":"queue","routing_key":"atguigu","arguments":{}},{"source":"exchange.fanout","vhost":"/","destination":"atguigu.emps","destination_type":"queue","routing_key":"atguigu.emps","arguments":{}},{"source":"exchange.fanout","vhost":"/","destination":"atguigu.news","destination_type":"queue","routing_key":"atguigu.news","arguments":{}},{"source":"exchange.fanout","vhost":"/","destination":"gulixueyuan.news","destination_type":"queue","routing_key":"gulixueyuan.news","arguments":{}},{"source":"exchange.topic","vhost":"/","destination":"atguigu.news","destination_type":"queue","routing_key":"*.news","arguments":{}},{"source":"exchange.topic","vhost":"/","destination":"gulixueyuan.news","destination_type":"queue","routing_key":"*.news","arguments":{}},{"source":"exchange.topic","vhost":"/","destination":"atguigu","destination_type":"queue","routing_key":"atguigu.#","arguments":{}},{"source":"exchange.topic","vhost":"/","destination":"atguigu.emps","destination_type":"queue","routing_key":"atguigu.#","arguments":{}},{"source":"my.exchange.direct","vhost":"/","destination":"atguigu","destination_type":"queue","routing_key":"atguigu","arguments":{}}]}

SpringBoot 整合 RabbitMQ

依赖&API

1. 引入spring-boot-starter-amqp（场景启动器）
2. 配置application.yml
3. 测试RabbitMQ
   1. AmqpAdmin：管理组件
   2. RabbitTemplate：消息发送处理组件

代码

依赖

<!-- amqp -->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件

# rabbit 配置信息
spring.rabbitmq.host=192.168.1.10
spring.rabbitmq.virtual-host=/
spring.rabbitmq.port=5672

api

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class GulimallOrderApplicationTests {@AutowiredAmqpAdmin amqpAdmin;@AutowiredRabbitTemplate rabbitTemplate;/*** 1、如何创建Exchange、Queue、Binding（AmqpAdmin）* 2、如何收发消息（RabbitTemplate）*/@Testpublic void contextLoads() {/*String name, boolean durable, boolean autoDelete, Map<String, Object> arguments*/DirectExchange directExchange = new DirectExchange("hello-java-exchange", true, false);amqpAdmin.declareExchange(directExchange);log.info("Exchange创建成功【{}】", directExchange.getName());}@Testpublic void testCreateQueue() {// exclusive true 就是只有一个消费者能连接Queue queue = new Queue("hello-java-queue", true, false, false);amqpAdmin.declareQueue(queue);log.info("Queue创建成功【{}】", queue.getName());}@Testpublic void testCreateBinding() {/*String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> arguments*/// 将exchange指定的交换机和destination目的地进行绑定，目的地的类型是什么，使用作为指定的路由键routingKeyBinding binding = new Binding("hello-java-queue", Binding.DestinationType.QUEUE, "hello-java-exchange", "hello-java-queue", null);amqpAdmin.declareBinding(binding);log.info("Binding创建成功");}@Testpublic void testSendMessage() {// 发送消息,如果发送的消息是个对象，我们会使用序列化机制，将对象写出去。对象必须实现Serializable接口String msg = "hello world";// 发送对象类型的消息 可以是json个格式的，自定义MessageConverter即可// rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", msg);for (int i = 0; i < 10; i++) {if (i % 2 == 0) {OrderReturnReasonEntity orderReturnReasonEntity = new OrderReturnReasonEntity();orderReturnReasonEntity.setId(1L);orderReturnReasonEntity.setCreateTime(new Date());orderReturnReasonEntity.setName("哈哈" + i);rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", orderReturnReasonEntity);}else{OrderEntity orderEntity = new OrderEntity();orderEntity.setOrderSn(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", orderEntity);}}log.info("消息发送完成：{}");}}

接受消息的注解

• 使用接听消息的注解，必须先启用 @EnableRabbit

监听消息：使用@RabbitListener,@RabbitHandler；必须有@EnableRabbit；* @RabbitListener: 类 + 方法（绑定消息，也可以直接放在方法上接受消息）
* @RabbitHandler：方法（重载区分不同的消息）

@Service("orderItemService")
@RabbitListener(queues = {"hello-java-queue"})
public class OrderItemServiceImpl{		/*** queues：声明需要监听的所有队列* org.springframework.amqp.core.Message* 参数可以有以下（都是可选参数，可以一个不行也可以写全参）* 1、Message message：原生消息详细信息。头+体* 2、T<发送的消息的类型> OrderReturnReasonEntity content* 3、Channel channel：当前传输数据的通道* * Queue：可以很多人都来监听。只要收到消息，队列删除消息，而且只能有一个人收到消息* 场景：* 1、 订单服务启动多个:同一个消息，只能有一个客户端能收到* 2、 只有一个消息完全处理完，方法运行结束，我们才可以接受下一个消息** @param message*/// @RabbitListener(queues = {"hello-java-queue"})@RabbitHandlerpublic void receiveMessage(Message message, OrderReturnReasonEntity content,Channel channel) throws InterruptedException {}@RabbitHandlerpublic void receiveMessage2(OrderEntity content) throws InterruptedException {}
}

自定义发送消息时的序列化器

@Configuration
public class MyRabbitConfig {@Beanpublic MessageConverter Jackson2JsonMessageConverter() {return new Jackson2JsonMessageConverter();}
}

自动配置原理

添加场景启动器spring-boot-starter-amqp

​ ==》导入RabbitAutoConfiguration

​ ==》注入属性文件@EnableConfigurationProperties(RabbitProperties.class)

​ ==》注入RabbitTemplate

​ ==》RabbitTemplate 有一个属性并有默认值private MessageConverter messageConverter = new SimpleMessageConverter();

​ ==> 在自动注入RabbitTemplate会从IOC容器中获取MessageConverter，获取得到就给属性重新赋值，实现替换消息的序列化器

​ ==》注入AmqpAdmin

RabbitMQ消息确认机制-可靠抵达

概述

可靠抵达-ConfirmCallback

• spring.rabbitmq.publisher-confirms=true
  • 在创建connectionFactory的时候设置PublisherConfirms(true)选项，开启confirmcallback。
  • CorrelationData：用来表示当前消息唯一性。
  • 消息只要被broker接收到就会执行confirmCallback，如果是cluster模式，需要所有broker接收到才会调用confirmCallback。
  • 被broker接收到只能表示message已经到达服务器，并不能保证消息一定会被投递到目标 queue里。所以需要用到接下来的returnCallback。
• 调用时机：broker成功收到生产者的消息

可靠抵达-returnCallback

• spring.rabbitmq.publisher-returns=true
• spring.rabbitmq.template.mandatory=true
  • confrim模式只能保证消息到达broker，不能保证消息准确投递到目标queue里。在有些业务场景下，我们需要保证消息一定要投递到目标queue里，此时就需要用到return退回模式。
  • 这样如果未能投递到目标queue里将调用returnCallback，可以记录下详细到投递数据，定期的巡检或者自动纠错都需要这些数据。
• 调用时机：比如我们发送的消息的route-key 匹配不到exchange下面的所有queue的时候会触发这个回调

可靠抵达一 Ack消息确认机制

• spring.rabbitmq.listener.simple.acknowledge-mode=manual

• 消费者获取到消息，成功处理，可以回复Ack给Broker

  • basic.ack用于肯定确认； broker将移除此消息
  • basic.nack用于否定确认；可以指定broker是否丢弃此消息，可以批量
  • basic.reject用于否定确认；同上，但不能批量

• 默认自动ack，消息被消费者收到，就会从broker的queue中移除

• queue无消费者，消息依然会被存储，直到消费者消费

• 消费者收到消息，默认会自动ack。但是如果无法确定此消息是否被处理完成，或者成功处理。我们可以开启手动ack模式

  • 消息处理成功，ack()， 接受下一个消息，此消息broker就会移除
  • 消息处理失败，nack()/reject()， 重新发送给其他人进行处理，或者容错处理后ack
  • 消息一直没有调用ack/nack方法， broker认为此消息正在被处理，不会投递给别人，此时客户端断开，消息不会被broker移除，会投递给别人

confirmCallback 和 returnCallback是发送端的回调，ack是消费端的回调。消息中间件成功接受生产者的消息会执行confirmCallback，交换机的信息为成功投递到queue会执行returnCallback，消费者成功收到消息会执行ack告诉broker将消息删除。

实操

配置文件

# rabbit 配置信息
spring.rabbitmq.host=192.168.1.10
spring.rabbitmq.virtual-host=/
spring.rabbitmq.port=5672# 开启发送端确认消息抵达Broker
spring.rabbitmq.publisher-confirms=true# 开启发送端消息抵达队列的确认
spring.rabbitmq.publisher-returns=true
# 只要消息不能抵达队列，会异步发送优先回调我们这个returnConfirm
spring.rabbitmq.template.mandatory=true# 手动ack消息（acknowledgement）（如果只想开启手动ack配置这个即可）
spring.rabbitmq.listener.simple.acknowledge-mode=manual

自定义RabbitTemplate

@Configuration
@EnableRabbit
public class MyRabbitConfig {@AutowiredRabbitTemplate rabbitTemplate;// 发送消息时的序列化机制@Beanpublic MessageConverter Jackson2JsonMessageConverter() {return new Jackson2JsonMessageConverter();}/*** 定制RabbitTemplate* 1、服务端收到消息就回调*  1、spring.rabbitmq.publisher-confirms=true*  2、设置确认回调ConfirmCallback** 2、消息没有正确抵达队列进行回调*  1、spring.rabbitmq.publisher-returns=true、spring.rabbitmq.template.mandatory=true*  2、设置失败回调ReturnCallback**  3、消费端确认（保证每个消息被正确消费，此时才可以broker删除这个消息）*  spring.rabbitmq.listener.simple.acknowledge-mode=manual 开启手动签收*  1、默认是自动确认的，只要消息接受到，消费端端会自动确认，服务端就会移除这个消息。*      问题：*          我们收到很多消息，自动回复给服务端ack。只有一个消息处理成功，宕机了。发生消息丢失。（通道一打开就回复ack）*          手动确认模式。只要我们没有明确告诉MQ，消息就一直是unacked转态。即使consumer宕机。队列里面的消息也不会丢失，consumer宕机则队列里面的消息会变为ready，*          下次有新的consumer连接进来就发给他（队列的消息变成unacked）。队列消息的转态是根据有没有消费者连接上队列*  2、如何签收：*       channel.basicAck(deliveryTag, false);签收：业务成功完成就应该签收*       channel.basicNack(deliveryTag,false,true); 拒签：业务失败，拒签。*/@PostConstruct // MyRabbitConfig 构造器创建完成以后，执行这个方法public void initRabbitTemplate() {// 设置确认回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 1、 只要消息抵达Broker就ack=true* @param correlationData 当前消息的唯一关联数据(这个是消息的唯一id)* @param ack 消息是否成功收到* @param cause 失败的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {System.out.println("confirm...correlationData[" + correlationData + "]==>ack[" + ack + "]==>[cause[" + cause + "]");}});// 设置消息没有抵达队列的失败回调rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** 只要消息没有投递给指定的队列，就触发这个失败回调* @param message 投递失败的消息详细信息* @param replyCode 回复的状态码* @param replyText 回复的文本内容* @param exchange 当时这个消息发给哪个交换机* @param routingKey 当时这个消息用哪个路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {System.out.println("Fail Message[" + message + "]==>replyCode[" + replyCode + "]==>replyText[" + replyText + "]==>exchange[" + exchange + "]==>routingKey[" + routingKey + "]");}});}
}

手动确认消息的api

void basicAck(long deliveryTag, boolean multiple) throws IOException; // 签收
void basicNack(long deliveryTag, boolean multiple, boolean requeue) throws IOException; // 拒签，可以批量
void basicReject(long deliveryTag, boolean requeue) throws IOException; // 拒签

@RabbitHandlerpublic void receiveMessage(Message message, OrderReturnReasonEntity content,Channel channel) throws InterruptedException {// {"id":1,"name":"哈哈","sort":null,"status":null,"createTime":1594892543985}System.out.println("接收到消息...内容：" + message + "===>内容：" + content);byte[] body = message.getBody();// 消息头属性信息MessageProperties messageProperties = message.getMessageProperties();
//        Thread.sleep(3000);System.out.println("消息处理完成=》" + content.getName());// channel内按顺序自增的long deliveryTag = message.getMessageProperties().getDeliveryTag();System.out.println("deliveryTag = " + deliveryTag);try {if (deliveryTag % 2 == 0) {//收货// 签收货物,非批量模式。批量模式就是将当前通道内里面的消息都ack，// 因为一个客户端只会与mq建立一条链接，客户端连接多个队列一个队列就是一个通道channel.basicAck(deliveryTag, false);System.out.println("签收了.." + deliveryTag);} else {// 退货// long deliveryTag, boolean multiple 批量处理 , boolean requeue 拒收之后是否重新发回队列// requeue=false 丢弃，requeue=true 发回服务器，服务器重新如入队 （重新入队的依据是我们的消息又发送了且在后面发送）
//                channel.basicReject(deliveryTag,true); // 不可批量channel.basicNack(deliveryTag,false,true); // 可批量System.out.println("没有签收了.." + deliveryTag);}} catch (Exception e) {// e.printStackTrace();// 网络中断} finally {}}

RabbitMQ延时队列（实现定时任务）

概述

场景：比如未付款订单，超过一定时间后，系统自动取消订单并释放占有物品。

常用解决方案：spring的schedule定时任务轮询数据库

缺点：消耗系统内存、增加了数据库的压力、存在较大的时间误差

解决： rabbitmq的消息TTL和死信Exchange结合

定时任务和rabbitmq延时队列的比较

延时队列的实现

延时队列的实现以一（推荐）：设置队列过期时间

流程：生产者发送消息到一个普通的交换机中，该交换机会将消息发送给一个设置过期时间的队列中。这个队列设置了ttl、路由键和死信交换机。当队列过期后，里面的消息就会发送到死信交换机中，然后死信交换机会根据消息的路由键发送给对应的队列。我们消费端就监听这个队列即可。

延时队列的实现以二：设置消息过期时间

流程：和上一个差不过。

总结

总结：推荐给队列设置过期时间实现延时队列。因为rabbitmq采用的是惰性检查机制，比如消息队列放入了3条消息，第一条是5分钟过过期、第二条是3分钟分钟过过期、第三条是1分钟过期。rabbitmq检测到第一个入队的消息过期时间是5分钟，就会等到5分钟之后在查看队列里面的消息。而不是消息一入队就检测。这就导致后进入的消息不能及时的过期。 所以最好的办法是设置队列有效期。

消息的TTL （Time To Live）

• 消息的TTL就是消息的存活时间。
• RabbitMQ可以对队列和消息分别设置TTL
  • 对队列设置就是队列没有消费者连着的保留时间，也可以对每一个单独的消息做单独的设置。超过了这个时间，我们认为这个消息就死了，称之为死信。
  • 如果队列设置了，消息也设置了，那么会取小的。所以一个消息如果被路由到不同的队列中，这个消息死亡的时间有可能不一样（不同的队列设置）。这里单讲单个消息的TTL，因为它才是实现延迟任务的关键。可以通过设置消息的expiration字段或者x-message-ttl属性来设置时间，两者是一样的效果

Dead Letter Exchanges （DLX）

• 一个消息在满足如下条件下，会进死信路由，记住这里是路由而不是队列，一个路由可以对应很多队列。（什么是死信）
  • 一个消息被Consumer拒收了，并且reject方法的参数里requeue是false。 也就是说不会
    被再次放在队列里，被其他消费者使用。（basic.reject/ basic.nack） requeue=false
  • 上面的消息的TTL到了，消息过期了。
  • 一队列的长度限制满了。排在前面的消息会被丢弃或者扔到死信路由上
• Dead Letter Exchange其实就是一种普通的exchange， 和创建其他exchange没有两样。只是在某一个设置Dead Letter Exchange的队列中有消息过期了，会自动触发消息的转发，发送到Dead Letter Exchange中去。
• 我们既可以控制消息在一段时间后变成死信，又可以控制变成死信的消息被路由到某一个指定的交换机， 结合二者，其实就可以实现一个延时队列

设计延时队列

总体设计思路

初步设计思路：两个交换机

最终设计方案：一个交换机

延时队列代码编写

流程

1. 在spring-boot环境下使用rabbitmq做定时任务（通过延时队列实现）

2. 导入依赖：

   <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
   </dependency>
   

3. 配置rabbitmq的主机地址、端口、虚拟主机地址

   spring.rabbitmq.host=192.168.1.10
   spring.rabbitmq.virtual-host=/
   spring.rabbitmq.port=5672
   

4. 通过@Bean的方式在rabbitmq中创建出exchange、queue、binding

代码

有效期队列的设置参数：x-dead-letter-exchange: order-event-exchange，x-dead-letter-routing-key: order.release.order，x-message-ttl: 60000（1000毫秒=1秒）

注意点：只有在连接上rabbitmq的时候我们@Bean注入的exchange、queue、binding才会在rabbitmq服务器上被创建出来。连接的方式可以采用@RabbitListener 通过消费者的方式建立一条连接。

@Configuration
public class MyMQConfig { @Beanpublic Queue orderDelayQueue() {// String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*x-dead-letter-exchange: order-event-exchangex-dead-letter-routing-key: order.release.order x-message-ttl: 60000*/Map<String, Object> arguments = new HashMap<>();arguments.put("x-dead-letter-exchange", "order-event-exchange");arguments.put("x-dead-letter-routing-key", "order.release.order");arguments.put("x-message-ttl", 60000);return new Queue("order.delay.queue", true, false, false, arguments);}@Beanpublic Exchange orderEventExchange() {// String name, boolean durable, boolean autoDelete, Map<String, Object> argumentsreturn new TopicExchange("order-event-exchange", true, false, null);}@Beanpublic Binding orderCreateOrderBinding() {// String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> argumentsreturn new Binding("order.delay.queue",Binding.DestinationType.QUEUE,"order-event-exchange","order.create.order",null);}
}

如何保证消息可靠性

消息丢失问题

• 消息发送出去，由于网络问题没有抵达服务器
  • 做好容错方法（trycatch） ，发送消息可能会网络失败，失败后要有重试机制，可记录到数据库，采用定期扫描重发的方式
  • 做好日志记录，每个消息状态是否都被服务器收到都应该记录
  • 做好定期重发，如果消息没有发送成功，定期去数据库扫描未成功的消息进行重发
• 消息抵达Broker， Broker要将消息写入磁盘（持久化）才算成功。避免出现Broker尚未持久化完成，宕机。
  • publisher必须加入确认回调机制，确认成功的消息，修改数据库消息状态。（修改RabbitTemplate）
  • 通过confirmCallback 确认broker是否持久化了数据
• 自动ACK的状态下。消费者收到消息，但没来得及消息然后宕机
  • 一定开启手动ACK，消费成功才移除，失败或者没来得及处理就noAck并重新入队

消息重复

• 消息消费成功，事务已经提交， ack时，机器宕机。导致没有ack成功，Broker的消息重新由unack变为ready，并发送给其他消费者
• 消息消费失败，由于重试机制，自动又将消息发送出去（这个是允许的）
• 成功消费，ack时宕机，消息由unack变为ready， Broker又重新发送
  • 消费者的业务消费接口应该设计为幂等性的。比如扣库存有工作单的状态标志
  • 使用防重表（redis/mysql） ，发送消息每一个都有业务的唯一 标识，处理过就不用处理
  • rabbitMQ的每一个消息都有redelivered字段，可以获取是否是被重新投递过来的，而不是第一次投递过来的

消息积压

• 消费者宕机积压，
• 消费者消费能力不足积压
• 发送者发送流量太大
  • 上线更多的消费者，进行正常消费
  • 上线专门的队列消费服务，将消息先批量取出来，记录数据库，离线慢慢处理

代码实现

消息丢失问题

// 1.业务逻辑 try catch
try {// TODO 保证消息一定会发送出去，每一个消息都可以做好日记记录（给数据库保存每一个消息的详细信息）// TODO 定期扫描数据库将失败的消息在发送一遍rabbitTemplate.convertAndSend("order-event-exchange", "stock.release.other", orderTo);} catch (Exception e) {// TODO 重新发送（意义不大，万一是服务器宕机了）// while
}// 2.建表进行日志记录
DROP TABLE IF EXISTS `mq_message`;
CREATE TABLE `mq_message` (`message_id` char(32) NOT NULL,`content` text,`to_exchane` varchar(255) DEFAULT NULL,`routing_key` varchar(255) DEFAULT NULL,`class_type` varchar(255) DEFAULT NULL,`message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',`create_time` datetime DEFAULT NULL,`update_time` datetime DEFAULT NULL,PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;// 3.确保服务器已经将消息持久化
# 开启发送端确认消息抵达Broker
spring.rabbitmq.publisher-confirms=true@Configuration
@EnableRabbit
public class MyRabbitConfig {@PostConstruct // MyRabbitConfig 构造器创建完成以后，执行这个方法public void initRabbitTemplate() {// 设置确认回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 1、 只要消息抵达Broker就ack=true* @param correlationData 当前消息的唯一关联数据(这个是消息的唯一id)* @param ack 消息是否成功收到* @param cause 失败的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {/*1、做好消息确认机制（publisher， consumer 【手动ack】）2、每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一遍*/// 服务器收到了// 修改消息的转态System.out.println("confirm...correlationData[" + correlationData + "]==>ack[" + ack + "]==>[cause[" + cause + "]");}});}// 4. 开启手动ack（使用 @RabbitListener、 @RabbitHandler() 需要@EnableRabbit）# 手动ack消息（acknowledgement）spring.rabbitmq.listener.simple.acknowledge-mode=manual@RabbitListener(queues = {"order.release.order.queue"})public class OrderCloseListener {@AutowiredOrderService orderService;@RabbitHandler()public void listener(Message message, OrderEntity entity, Channel channel) throws IOException {System.out.println("收到过期的订单信息，准备关闭订单" + entity.toString());try {orderService.closeOrder(entity);// 手动签收消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 消息消费失败，重新入队channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);}}
}

消息重复问题

// 1.业务操作设计成幂等性（多次操作结果是一样的）
// 先确定记录的转态，在修改
wareOrderTaskDetailService.list(new QueryWrapper<WareOrderTaskDetailEntity>().eq("task_id", id).eq("lock_status", 1));
update wms_ware_sku set stock_locked = stock_locked - #{num} where sku_id = #{skuId} and ware_id = #{wareId};// 2. 利用rabbitmq message的属性判断
Boolean redelivered = message.getMessageProperties().getRedelivered(); // 当前消息是否被第二次及以后（重新）派发过来

消息挤压：多开服务器呗

总结

1. 最重要的问题是防止消息丢失

   1. 做好消息确认机制（publisher， consumer 【手动ack】）
   2. 每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一遍

2. 消息重复问题：将业务设计成幂等操作（最简单就这么做）

3. 消息挤压：多开服务器呗

最终代码

依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

配置文件

# ==== rabbit start ======
# rabbit 配置信息
spring.rabbitmq.host=192.168.1.10
spring.rabbitmq.virtual-host=/
spring.rabbitmq.port=5672# 开启发送端确认消息抵达Broker
spring.rabbitmq.publisher-confirms=true# 开启发送端消息抵达队列的确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达队列，会异步发送优先回调我们这个returnConfirm
spring.rabbitmq.template.mandatory=true# 手动ack消息（acknowledgement）
spring.rabbitmq.listener.simple.acknowledge-mode=manual
# ==== rabbit end ======

配置类(初始化rabbitTemplate)

package com.atguigu.gulimall.order.config;import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import javax.annotation.PostConstruct;/*** @author: haitao* @email: haitaoss@aliyun.com* @date: 2020/7/16 17:25:57*/
@Configuration
@EnableRabbit
public class MyRabbitConfig {@AutowiredRabbitTemplate rabbitTemplate;@Beanpublic MessageConverter Jackson2JsonMessageConverter() {return new Jackson2JsonMessageConverter();}/*** 定制RabbitTemplate* 1、服务端收到消息就回调*  1、spring.rabbitmq.publisher-confirms=true*  2、设置确认回调ConfirmCallback** 2、消息没有正确抵达队列进行回调*  1、spring.rabbitmq.publisher-returns=true、spring.rabbitmq.template.mandatory=true*  2、设置失败回调ReturnCallback**  3、消费端确认（保证每个消息被正确消费，此时才可以broker删除这个消息）*  spring.rabbitmq.listener.simple.acknowledge-mode=manual 开启手动签收*  1、默认是自动确认的，只要消息接受到，消费端端会自动确认，服务端就会移除这个消息。*      问题：*          我们收到很多消息，自动回复给服务端ack。只有一个消息处理成功，宕机了。发生消息丢失。（通道一打开就回复ack）*          手动确认模式。只要我们没有明确告诉MQ，消息就一直是unacked转态。即使consumer宕机。队列里面的消息也不会丢失，consumer宕机则队列里面的消息会变为ready，*          下次有新的consumer连接进来就发给他（队列的消息变成unacked）。队列消息的转态是根据有没有消费者连接上队列*  2、如何签收：*       channel.basicAck(deliveryTag, false);签收：业务成功完成就应该签收*       channel.basicNack(deliveryTag,false,true); 拒签：业务失败，拒签。*/@PostConstruct // MyRabbitConfig 构造器创建完成以后，执行这个方法public void initRabbitTemplate() {// 设置确认回调rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {/*** 1、 只要消息抵达Broker就ack=true* @param correlationData 当前消息的唯一关联数据(这个是消息的唯一id)* @param ack 消息是否成功收到* @param cause 失败的原因*/@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {/*1、做好消息确认机制（publisher， consumer 【手动ack】）2、每一个发送的消息都在数据库做好记录。定期将失败的消息再次发送一遍*/// 服务器收到了// 修改消息的转态System.out.println("confirm...correlationData[" + correlationData + "]==>ack[" + ack + "]==>[cause[" + cause + "]");}});// 设置消息没有抵达队列的失败回调rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {/*** 只要消息没有投递给指定的队列，就触发这个失败回调* @param message 投递失败的消息详细信息* @param replyCode 回复的状态码* @param replyText 回复的文本内容* @param exchange 当时这个消息发给哪个交换机* @param routingKey 当时这个消息用哪个路由键*/@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {//报错误了 。修改数据库当前消息的状态->错误。System.out.println("Fail Message[" + message + "]==>replyCode[" + replyCode + "]==>replyText[" + replyText + "]==>exchange[" + exchange + "]==>routingKey[" + routingKey + "]");}});}
}

通过@Bean创建exchange、queue、binding

package com.atguigu.gulimall.order.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;import java.util.HashMap;
import java.util.Map;/*** @author: haitao* @email: haitaoss@aliyun.com* @date: 2020/7/21 15:36:58* @Description*/
@Configuration
public class MyMQConfig {// spring boot 支持通过@Bean的方式创建出Exchange、Queue、Binding。他会自动使用amqpAdmin帮我创建到指定的消息中间件中/*** 容器中的Binding， Queue， Exchange 都会自动创建（RabbitMQ没有的情况）* RabbitMQ只要有。@Bean声明的属性发生变化也不会覆盖** @return*/@Beanpublic Queue orderDelayQueue() {// String name, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments/*x-dead-letter-exchange: order-event-exchangex-dead-letter-routing-key: order.release.orderx-message-ttl: 60000*/Map<String, Object> arguments = new HashMap<>();arguments.put("x-dead-letter-exchange", "order-event-exchange");arguments.put("x-dead-letter-routing-key", "order.release.order");arguments.put("x-message-ttl", 60000);return new Queue("order.delay.queue", true, false, false, arguments);}@Beanpublic Queue orderReleaseQueue() {return new Queue("order.release.order.queue", true, false, false, null);}@Beanpublic Exchange orderEventExchange() {// String name, boolean durable, boolean autoDelete, Map<String, Object> argumentsreturn new TopicExchange("order-event-exchange", true, false, null);}@Beanpublic Binding orderCreateOrderBinding() {// String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> argumentsreturn new Binding("order.delay.queue",Binding.DestinationType.QUEUE,"order-event-exchange","order.create.order",null);}@Beanpublic Binding orderReleaseOrderBinding() {// String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> argumentsreturn new Binding("order.release.order.queue",Binding.DestinationType.QUEUE,"order-event-exchange","order.release.order",null);}/*** 订单释放直接和库存释放进行绑定* （因为订单解锁消息的发送可能由于网络问题，比库存解锁消息发出的慢。导致库存释放解锁了。* 库存释放解锁逻辑是获取订单的转态不是创建就解锁库存，然后只要不报错就把消息消费掉。）** @return*/@Beanpublic Binding orderReleaseOtherBinding() {// String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> argumentsreturn new Binding("stock.release.stock.queue",Binding.DestinationType.QUEUE,"order-event-exchange","stock.release.other.#",null);}
}

通过api创建exchange、queue、binding

public class GulimallOrderApplicationTests {@AutowiredAmqpAdmin amqpAdmin;@AutowiredRabbitTemplate rabbitTemplate;/*** 1、如何创建Exchange、Queue、Binding（AmqpAdmin）* 2、如何收发消息（RabbitTemplate）*/@Testpublic void contextLoads() {/*String name, boolean durable, boolean autoDelete, Map<String, Object> arguments*/DirectExchange directExchange = new DirectExchange("hello-java-exchange", true, false);amqpAdmin.declareExchange(directExchange);log.info("Exchange创建成功【{}】", directExchange.getName());}@Testpublic void testCreateQueue() {// exclusive true 就是只有一个消费者能连接Queue queue = new Queue("hello-java-queue", true, false, false);amqpAdmin.declareQueue(queue);log.info("Queue创建成功【{}】", queue.getName());}@Testpublic void testCreateBinding() {/*String destination, DestinationType destinationType, String exchange, String routingKey,Map<String, Object> arguments*/// 将exchange指定的交换机和destination目的地进行绑定，目的地的类型是什么，使用作为指定的路由键routingKeyBinding binding = new Binding("hello-java-queue", Binding.DestinationType.QUEUE, "hello-java-exchange", "hello-java-queue", null);amqpAdmin.declareBinding(binding);log.info("Binding创建成功");}@Testpublic void testSendMessage() {// 发送消息,如果发送的消息是个对象，我们会使用序列化机制，将对象写出去。对象必须实现Serializable接口String msg = "hello world";// 发送对象类型的消息 可以是json个格式的，自定义MessageConverter即可// rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", msg);for (int i = 0; i < 10; i++) {if (i % 2 == 0) {OrderReturnReasonEntity orderReturnReasonEntity = new OrderReturnReasonEntity();orderReturnReasonEntity.setId(1L);orderReturnReasonEntity.setCreateTime(new Date());orderReturnReasonEntity.setName("哈哈" + i);rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", orderReturnReasonEntity);}else{OrderEntity orderEntity = new OrderEntity();orderEntity.setOrderSn(UUID.randomUUID().toString());rabbitTemplate.convertAndSend("hello-java-exchange", "hello-java-queue", orderEntity);}}log.info("消息发送完成：{}");}}

建表解决消息的丢失问题

// 2.建表进行日志记录
DROP TABLE IF EXISTS `mq_message`;
CREATE TABLE `mq_message` (`message_id` char(32) NOT NULL,`content` text,`to_exchane` varchar(255) DEFAULT NULL,`routing_key` varchar(255) DEFAULT NULL,`class_type` varchar(255) DEFAULT NULL,`message_status` int(1) DEFAULT '0' COMMENT '0-新建 1-已发送 2-错误抵达 3-已抵达',`create_time` datetime DEFAULT NULL,`update_time` datetime DEFAULT NULL,PRIMARY KEY (`message_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

监听队列获取消息

@Service
@RabbitListener(queues = {"order.release.order.queue"})
public class OrderCloseListener {@AutowiredOrderService orderService;@RabbitHandler()public void listener(Message message, OrderEntity entity, Channel channel) throws IOException {System.out.println("收到过期的订单信息，准备关闭订单" + entity.toString());try {orderService.closeOrder(entity);// 手动签收消息channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);} catch (Exception e) {// 消息消费失败，重新入队channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);}}
}