背景前面我们已经学会了MQ的基本使用，但是我们真正到企业实际开发中，消息业务难免会出现问题，问题的出现主要再这下面三个方面：1.消息没发到消息代理2.消息代理把消息弄丢了3.消费者把消息弄丢了因此我们如果要解决这样的消息可靠性的问题，我们就要在这三个方面下功夫去优化：1.发送者可靠性、2.MQ(消息代理)可靠性、3.消费者可靠性我们做完这三个优化后可以大大降低消息可靠性的问题，但是还是会有小概率发生，因此，我们还需要相关的兜底方案发送者可靠性发送者重连发送者重连用于解决由于网络波动，导致发送者连接MQ失败的情况，因此我们可以修改配置文件让发送者重新连接MQ各个参数在下图写的非常清楚了，这里就不多赘述了我们修改自己发送方的配置文件，关闭rabbitMQ，直接运行我们发送消息的测试方法看看会是什么效果结果如下：可以看到，发送消息失败了，并且按照我们的配置文件所配置的，一共连接了3次，且每一次都是间隔1*1秒的等待时间注意：发送者重连机制确实可以一定的解决该问题，但是该机制是阻塞式的，若“发送消息”后面还有其他代码，那么其他代码必须要等待我们重连完成后才会执行，因此这种阻塞式的代码会影响其他业务的性能，从而影响用户体验。所以我们要合理配置等待时长后重试次数发送者确认介绍AMQP提供了PublisherConfirm和PublishReturn两种机制共同来确认判断消息在MQ的处理情况，如果在MQ处理的没问题，那就返回正常结果，反之返回异常结果。如果出现了异常结果，我们就可以尝试去做重连以及其他的一些异常处理注意：发送者确认机制由于需要和MQ进行通信和确认，因此会大大影响到发送消息的效率，并且发送者和MQ连接这一块出现的异常比较少，因此真实企业开发不是特别推荐开启返回结果介绍：ACK：消息在MQ中处理正常所返回的结果，情况可以参照上图。这里补充一点，上图中第一条ACK情况：投递成功但是“路由失败”返回ACK，这里你可能会想，明明都“失败”了，为什么还有返回ACK呢？因为ACK返回的条件是MQ本身处理是否正常，如果路由异常，比如上图的exchange2交换机没有连接任何的队列，向他发送消息，必定路由失败出现异常，那么交换机没有连接队列是谁造成的呢？必然不是MQ本身，而是我们程序员没有配置好MQ的交换机或者是在用代码生成交换机和队列的时候没有写好所造成的，这种错误不来源于MQ本身，来源于外部(比如作为程序员的我们)，因此返回的是ACK，同理还有其他类似情况如routingkey配置错误导致路由失败的，同样返回的是ACKNACK：就是投递失败出现的异常结果，这就是MQ本身出现问题了，比如说“持久消息”投递到了MQ，但是并没有完成持久化，此时就会返回NACK介绍完基础部分我们来介绍如何开启“发送者确认”机制吧启动步骤1.文件配置修改相关yml文件，开启我们前文所说实现“发送者确认”的PublisherConfirm和PublishReturn两种机制，其中PublisherConfirm有三种模式，simple就是我们前文有所提及的“阻塞式”的同步执行，会影响业务性能，而correlated是异步的方式，因此我们更加推荐使用correlated方式启动2.PublishReturn机制开启在publisher新建config包，在里面新建配置类即可3.PublisherConfirm机制开启使用该机制发送消息的代码如下图，仔细一看图片，最下面一行不就是我们之前写发送消息的代码吗，只不过多了一个“cd”参数，因此我们到时候优化修改代码就是优化我们之前写的发消息的代码，接下来我们一一简单介绍：首先是多了的cd参数，该参数是一个对象，这个对象中包含了两个重要的属性：1.id：用于标识每一个消息，将来消息很多，需要通过id才知道哪一个消息出现问题2.future：异步得到确认结果的关键关于上文所讲的future，这里可以简单介绍一下，我们之前编写yml文件使用了correlated即异步回调方式返回确认结果，这里的future就是用于实现异步回调的确认结果的，对于确认结果我们不需要等待他返回才去执行下面的代码，我们可以直接返回一个future对象，然后什么时候确认结果返回了，就可以利用future对象执行回调方法得到确认结果。(下图标红框的代码可以参照)剩余代码onFailure/onSuccess就是用于判断MQ处理是否正常，其中onfailure不用特别注意，基本不会触发，ACK/NACK的判断在onSuccess方法最后我们可以看发送者发送消息到MQ的成功与否了MQ可靠性MQ默认接收到消息是存储在MQ的内存当中的，因为这样无论消息的读或写速度都比较快，但是这也带来一个问题，当大量的消息发送到MQ中，MQ的内存队列直接满了，此时如果还想要接收消息，MQ就会“被动的”把之前比较老的消息写入磁盘中，这样就可以给内存队列腾出空间从而又可以接收新的信息了，但是“写”的动作是非常耗时的，因此在“写”的过程中，队列可能依旧是满的状态，此时外部新发来的消息就会丢失为了解决这个问题，就要在MQ上做处理数据持久化简单来说就是要把exchange交换机、队列queue、发送的消息message让其的类型都变为durable持久化的形式，但是由于我们使用SpringAMQP发送的，因此默认来说全部都是持久化，我们无需做任何的操作，在这里我们就不过多演示赘述了LazyQueue这里也是简单说一下，因为从3.12版本开始，RabbitMQ的队列默认都是LazyQueue，因此这一块优化官方以及帮我们做好了，我们不用怎么管，只不过我们这次学习用的还是3.8，因此简单了解一下即可对于非LazyQueue的“传统队列”，面对高并发的消息非常难处理，因此这时就要开启LazyQueue开启的方式也很简单，对于控制台，直接选择“lazymode”即可对于代码，和以前生成交换机和队列形式一样，使用声明bean对象或者注解的方式即可，这里不多赘述了消费者可靠性消费者确认机制前文我们讲过了发送者确认机制，消费者确认机制也是同理，通过消费者是否正确处理消息所返回的确认结果来做异常的处理这里我们需要重点提醒的是，返回确认结果的时机，一定是消息发送到消费者，并且消费者处理完成后这些消息才返回确认结果，不能是消息一发送给消费者就返回确认结果，这样确认消息就没有什么意义了确认结果：ACK/NACK：和消费者确认的确认结果一样，只不过他们新增了机制，ACK的返回代表了消息被消费者成功处理，并且这个成功处理的消息在处理完成后要被删除(很好理解，比较都处理完成了，就不要留在队列里面占用内存了)。NACK的返回代表消息处理失败，并且这个失败可能是消费者没有处理好等其他问题，因此需要重新给消费者发一次消息，直到消费者处理成功即可。reject：也是消息处理失败的返回结果，只不过这个“失败”的原因并不来源于消费者，更多的是来源于消息本身，可能消息的内容、格式本身就是错误的，因此发送reject结果直接拒绝处理该消息，因此该消息直接会被RabbitMQ删除。(毕竟如果消息本身就是错误的，重新发给消费者处理再多次都没有用)开启启动也很简单，只需要再yml文件修改配置即可一共有三种状态，none就是我们之前讲过的消息一发送到消费者就立马返回结果，这是十分不推荐的，manuel手动模式，可以自定义再任意时刻返回确认信息，比较灵活但是需要自己手写代码，一般是定制化需求会用，这里不做演示。auto自动模式，就是我们上文讲解的逻辑，在“合适的时候”(上文讲的返回情况)返回相应的确认结果消费者重试我们前文讲的消费者确认机制，当返回NACK时，会把消息再次发送给消费者做消息处理，直到成功位置。但是如果出现一些极端情况，比如这个错误一直没有得到处理，那么这个消息就会一直发送给消费者处理，一直重复循环，而消息的发送量是比较大的，一秒内发送一千多条，那么这样就会占用不必要的资源，因此我们使用失败重试机制，当消费者处理消息出现异常时，先不是一直发消息给消费者，而是在本地进行重试，再一条条的发送给消费者试，直到成功为止(这个需要配置)，这里再发送给消费者，消息流量就会小特别多，可能几秒钟才几条基本配置和我们之前的发送者重连的配置差不多，这里不多赘述，指的一提的就是stateless这个配置，就是你的业务中是否包含了事务，如果包含了就填写false，不包含就填写true我们配置好后，主动抛出一个异常，看看消费者重试默认时如何处理的可以看到，按照了我们配置文件的配置方式本地重试了三次，但是我们发现一个问题，重试完后消息直接就被“reject”丢掉了，那么这条消息就变相的丢失了，这明显不符合需求因此，再消费者重试机制我们还需要再次配置其他的策略刚刚的reject策略是默认的策略，直接丢弃消息，更加推荐的是下图中的下面两种策略，其中ImmeadiateRequeueMessageRecover就是我们之前说的“慢慢的一条条重新入队给消费者处理”，另一个RepublishMessageRecover是一种方案，当重试耗尽后，把失败消息发送给指定的交换机，这个交换机再发送给指定的队列，这样我们开发人员就可以从队列获取到错误信息，这样子开发人员就可以介入从而处理异常我们下面来演示将失败消息投递到指定交换机RepublishMessageRecover方案实例我们新建配置类，做好错误交换机/队列以及失败处理策略的配置然后我们再来测试错误：控制台成功显示把错误信息发送给特定交换机了来到rabbitMQ控制台，可以成功看到错误消息以及错误原因，非常方便程序员介入处理业务幂等性幂等性用在我们软件业务上也很好理解，一个业务执行多次它的结果都是一样的，那么这个业务就是幂等的，比如查询商品就是幂等的，扣减库存就不是幂等的，因为多次执行库存就会一直减少了解完幂等的概念，我们MQ对于非幂等的业务这里还会出现部分差错，比如：MQ给消费者发送“扣减库存”的消息，消费者执行完成后，返回ACK给MQ，但是此时，MQ和消费者之间的网络断开了，导致消费者的ACK没有发送给MQ，MQ就会认为消费者宕机了没有收到它发送的消息，于是又会准备发“扣减库存”的消息给消费者，当网络恢复时，MQ再次向“以为宕机”的消费者发送扣减库存的消息，消费者再次接收消息，执行相关服务，于是库存就又被扣减了对于非幂等的业务，我们有两种主要的处理方案唯一消息ID给MQ里面的消息都“设置”唯一ID，消费者接收消息后处理好业务把消息ID保存到数据库，如果下次又收到了相同消息，只需要去数据库查看该消息ID是否存在，若存在则放弃处理这个ID时MQ消息的一种属性，我们可以这样设置：在我们上一篇文章末尾介绍的消息转换器，把所发送的消息转换为MQ专属的“Message”类型对象，ID就是Message对象中的一个属性，因此我们在消息转换器中打开相关设置，然后消费者接收消息的时候去get message对象以及它的ID属性即可因此，我们在发送方的启动类修改一下我们之前定义的消息转换器，把自动创建消息id的值改为true我们设置完尝试发送消息，发现队列中的消息成功附赠id属性发送方设置完了，我们消费者接收也得接收相应类型的参数，我们前文提过的Message对象消费者也能成功接收，并且从对象中提取出相关属性业务判断这个其实就是对于特定的业务做相关的优化了，不算很泛的一种方式，这里就简单讲一下吧，实际开发中需要更具特定的业务来思考要怎么样优化场景：支付服务和交易服务，用户点击支付，交易服务吧订单标记为已支付，但是此时MQ和消费者之间的网络断开了，没有返回ACK，因此MQ又准备发送同样的“标记支付”消息。但是此时用户又想退款了，并且此时MQ和消费者之间的网络又好了，那么先前的“标记支付”的消息就会直接覆盖掉用户这次“退款消息”解决方案也很简单：消费者接收到消息去执行相关订单状态操作的业务时，去查看当前订单状态，若当前订单状态和现在一样，则不执行该业务/不管该消息，比如现在就是已支付状态，又让我干一次“已支付”那就不干了延迟消息设想一个场景：假设创建一个订单，首先通过“交易服务”创建保存订单信息，再通过“商品服务”来扣减相关商品的库存，同时调用“支付服务”来更新订单状态(是否支付)。此时如果有某一些不可抗力的因素导致“支付服务”的返回结果无法返回给交易服务，导致订单的状态一直是“未支付”，这就会导致很大的问题，假设用户已经支付了，但是又显示未支付，体验非常差。假设用户没支付，但是前面调用“商品服务”的时候，商品库存已经减少了，导致别的用户无法购买该商品，同样体验非常差因此我们可以使用延迟消息来解决该问题交易服务创建完订单后可以设置一个“闹钟”，让这个“闹钟”再设定时间段叫醒自己去查看支付服务的的返回结果，若支付服务有相关返回结果，那么就更新订单状态，如果没有返回结果，这说明“支付超时”，则取消订单，复原商品的库存延迟消息就是用来充当这个“闹钟”的，交易服务通过发送延迟消息到MQ，指定时间后发送消息给交易服务，让交易服务去查询支付服务的返回结果那我们要如何实现“延迟消息”呢，我们的rabbitMQ本身并不具备“延迟消息”的功能，但是我们可以通过里面其他属性去“模拟”延迟消息从而达到我们所想要的效果死信交换机顾名思义，就是由“死信”和“交换机”共同组成的先来了解一下什么是“死信”(如下图所示)，其中对于实现“延迟消息”最重要的就是其中的第二条：如果消息是一个过期消息，并且超时无人消费，那么该消息就会变成“死信”了解完这个特性后，我们通过死信交换机来实现延迟消息就非常容易了：我们先创建一个普通的交换机和队列，但是这个队列不绑定任何的消费者(重点!)，这样我们如果发送一些定期超时消息，由于没有人消费，留在队列里面的消息就会变成死信，然后再在该队列中绑定另一组队列和交换机(注意：这里的绑定需要使用dead-letter-exchange=绑定的交换机名称，来绑定，这样子队列中的死信就会进入所绑定的交换机)，这一组交换机和队列就会接收所绑定的队列发来的死信，最后再发送给消费者我们举例子来看，假设我们发一个30s过期的消息到第一组交换机和队列中，由于第一组交换机队列没有任何的消费者，那么该消息等到30s后会自动变为死信，而死信就会接着进入我们绑定的死信交换机队列再发送给消费者。我们去除过程只看结果，我们发送的定期过期消息不就变成了”延迟消息“了吗，定期过期多久就代表了延迟多久发送到消费者实践接下来我们来试着实践一下，其实思路很简单，无非就是创建两组队列和交换机，然后用特定的“死信”绑定方式来绑定第一组队列和第二组死信交换机即可首先先来声明死信交换机以及它的队列和消费者：再来声明绑定死信交换机的普通队列和交换机，由于普通队列不能有消费者，因此只能用bean对象的方式声明，不能像上面死信交换机一样使用注解的方式声明要注意，普通队列绑定其死信交换机所用到的key要和死信交换机绑定它的死信队列所用的key保持完全一致，这里可以看到都是“dead”发送延迟消息最后我们来看测试结果：消息在31秒的时候发出消费者在36秒的时候才接收到完全符合我们使用死信交换机实现延迟消息延迟消息插件上文讲解了死信交换实现延迟消息，但是过于麻烦你，社区中已经出现了专门设置交换机为“延迟消息交换机”的插件，从而实现延迟消息使用也非常简单，就是在交换机加一个delay属性即可声明好消费者最后编写好测试方法可以看到成功实现了延迟消息
背景前面我们已经学会了MQ的基本使用，但是我们真正到企业实际开发中，消息业务难免会出现问题，问题的出现主要再这下面三个方面：1.消息没发到消息代理2.消息代理把消息弄丢了3.消费者把消息弄丢了因此我们如果要解决这样的消息可靠性的问题，我们就要在这三个方面下功夫去优化：1.发送者可靠性、2.MQ(消息代理)可靠性、3.消费者可靠性我们做完这三个优化后可以大大降低消息可靠性的问题，但是还是会有小概率发生，因此，我们还需要相关的兜底方案发送者可靠性发送者重连发送者重连用于解决由于网络波动，导致发送者连接MQ失败的情况，因此我们可以修改配置文件让发送者重新连接MQ各个参数在下图写的非常清楚了，这里就不多赘述了我们修改自己发送方的配置文件，关闭rabbitMQ，直接运行我们发送消息的测试方法看看会是什么效果结果如下：可以看到，发送消息失败了，并且按照我们的配置文件所配置的，一共连接了3次，且每一次都是间隔1*1秒的等待时间注意：发送者重连机制确实可以一定的解决该问题，但是该机制是阻塞式的，若“发送消息”后面还有其他代码，那么其他代码必须要等待我们重连完成后才会执行，因此这种阻塞式的代码会影响其他业务的性能，从而影响用户体验。所以我们要合理配置等待时长后重试次数发送者确认介绍AMQP提供了PublisherConfirm和PublishReturn两种机制共同来确认判断消息在MQ的处理情况，如果在MQ处理的没问题，那就返回正常结果，反之返回异常结果。如果出现了异常结果，我们就可以尝试去做重连以及其他的一些异常处理注意：发送者确认机制由于需要和MQ进行通信和确认，因此会大大影响到发送消息的效率，并且发送者和MQ连接这一块出现的异常比较少，因此真实企业开发不是特别推荐开启返回结果介绍：ACK：消息在MQ中处理正常所返回的结果，情况可以参照上图。这里补充一点，上图中第一条ACK情况：投递成功但是“路由失败”返回ACK，这里你可能会想，明明都“失败”了，为什么还有返回ACK呢？因为ACK返回的条件是MQ本身处理是否正常，如果路由异常，比如上图的exchange2交换机没有连接任何的队列，向他发送消息，必定路由失败出现异常，那么交换机没有连接队列是谁造成的呢？必然不是MQ本身，而是我们程序员没有配置好MQ的交换机或者是在用代码生成交换机和队列的时候没有写好所造成的，这种错误不来源于MQ本身，来源于外部(比如作为程序员的我们)，因此返回的是ACK，同理还有其他类似情况如routingkey配置错误导致路由失败的，同样返回的是ACKNACK：就是投递失败出现的异常结果，这就是MQ本身出现问题了，比如说“持久消息”投递到了MQ，但是并没有完成持久化，此时就会返回NACK介绍完基础部分我们来介绍如何开启“发送者确认”机制吧启动步骤1.文件配置修改相关yml文件，开启我们前文所说实现“发送者确认”的PublisherConfirm和PublishReturn两种机制，其中PublisherConfirm有三种模式，simple就是我们前文有所提及的“阻塞式”的同步执行，会影响业务性能，而correlated是异步的方式，因此我们更加推荐使用correlated方式启动2.PublishReturn机制开启在publisher新建config包，在里面新建配置类即可3.PublisherConfirm机制开启使用该机制发送消息的代码如下图，仔细一看图片，最下面一行不就是我们之前写发送消息的代码吗，只不过多了一个“cd”参数，因此我们到时候优化修改代码就是优化我们之前写的发消息的代码，接下来我们一一简单介绍：首先是多了的cd参数，该参数是一个对象，这个对象中包含了两个重要的属性：1.id：用于标识每一个消息，将来消息很多，需要通过id才知道哪一个消息出现问题2.future：异步得到确认结果的关键关于上文所讲的future，这里可以简单介绍一下，我们之前编写yml文件使用了correlated即异步回调方式返回确认结果，这里的future就是用于实现异步回调的确认结果的，对于确认结果我们不需要等待他返回才去执行下面的代码，我们可以直接返回一个future对象，然后什么时候确认结果返回了，就可以利用future对象执行回调方法得到确认结果。(下图标红框的代码可以参照)剩余代码onFailure/onSuccess就是用于判断MQ处理是否正常，其中onfailure不用特别注意，基本不会触发，ACK/NACK的判断在onSuccess方法最后我们可以看发送者发送消息到MQ的成功与否了MQ可靠性MQ默认接收到消息是存储在MQ的内存当中的，因为这样无论消息的读或写速度都比较快，但是这也带来一个问题，当大量的消息发送到MQ中，MQ的内存队列直接满了，此时如果还想要接收消息，MQ就会“被动的”把之前比较老的消息写入磁盘中，这样就可以给内存队列腾出空间从而又可以接收新的信息了，但是“写”的动作是非常耗时的，因此在“写”的过程中，队列可能依旧是满的状态，此时外部新发来的消息就会丢失为了解决这个问题，就要在MQ上做处理数据持久化简单来说就是要把exchange交换机、队列queue、发送的消息message让其的类型都变为durable持久化的形式，但是由于我们使用SpringAMQP发送的，因此默认来说全部都是持久化，我们无需做任何的操作，在这里我们就不过多演示赘述了LazyQueue这里也是简单说一下，因为从3.12版本开始，RabbitMQ的队列默认都是LazyQueue，因此这一块优化官方以及帮我们做好了，我们不用怎么管，只不过我们这次学习用的还是3.8，因此简单了解一下即可对于非LazyQueue的“传统队列”，面对高并发的消息非常难处理，因此这时就要开启LazyQueue开启的方式也很简单，对于控制台，直接选择“lazymode”即可对于代码，和以前生成交换机和队列形式一样，使用声明bean对象或者注解的方式即可，这里不多赘述了消费者可靠性消费者确认机制前文我们讲过了发送者确认机制，消费者确认机制也是同理，通过消费者是否正确处理消息所返回的确认结果来做异常的处理这里我们需要重点提醒的是，返回确认结果的时机，一定是消息发送到消费者，并且消费者处理完成后这些消息才返回确认结果，不能是消息一发送给消费者就返回确认结果，这样确认消息就没有什么意义了确认结果：ACK/NACK：和消费者确认的确认结果一样，只不过他们新增了机制，ACK的返回代表了消息被消费者成功处理，并且这个成功处理的消息在处理完成后要被删除(很好理解，比较都处理完成了，就不要留在队列里面占用内存了)。NACK的返回代表消息处理失败，并且这个失败可能是消费者没有处理好等其他问题，因此需要重新给消费者发一次消息，直到消费者处理成功即可。reject：也是消息处理失败的返回结果，只不过这个“失败”的原因并不来源于消费者，更多的是来源于消息本身，可能消息的内容、格式本身就是错误的，因此发送reject结果直接拒绝处理该消息，因此该消息直接会被RabbitMQ删除。(毕竟如果消息本身就是错误的，重新发给消费者处理再多次都没有用)开启启动也很简单，只需要再yml文件修改配置即可一共有三种状态，none就是我们之前讲过的消息一发送到消费者就立马返回结果，这是十分不推荐的，manuel手动模式，可以自定义再任意时刻返回确认信息，比较灵活但是需要自己手写代码，一般是定制化需求会用，这里不做演示。auto自动模式，就是我们上文讲解的逻辑，在“合适的时候”(上文讲的返回情况)返回相应的确认结果消费者重试我们前文讲的消费者确认机制，当返回NACK时，会把消息再次发送给消费者做消息处理，直到成功位置。但是如果出现一些极端情况，比如这个错误一直没有得到处理，那么这个消息就会一直发送给消费者处理，一直重复循环，而消息的发送量是比较大的，一秒内发送一千多条，那么这样就会占用不必要的资源，因此我们使用失败重试机制，当消费者处理消息出现异常时，先不是一直发消息给消费者，而是在本地进行重试，再一条条的发送给消费者试，直到成功为止(这个需要配置)，这里再发送给消费者，消息流量就会小特别多，可能几秒钟才几条基本配置和我们之前的发送者重连的配置差不多，这里不多赘述，指的一提的就是stateless这个配置，就是你的业务中是否包含了事务，如果包含了就填写false，不包含就填写true我们配置好后，主动抛出一个异常，看看消费者重试默认时如何处理的可以看到，按照了我们配置文件的配置方式本地重试了三次，但是我们发现一个问题，重试完后消息直接就被“reject”丢掉了，那么这条消息就变相的丢失了，这明显不符合需求因此，再消费者重试机制我们还需要再次配置其他的策略刚刚的reject策略是默认的策略，直接丢弃消息，更加推荐的是下图中的下面两种策略，其中ImmeadiateRequeueMessageRecover就是我们之前说的“慢慢的一条条重新入队给消费者处理”，另一个RepublishMessageRecover是一种方案，当重试耗尽后，把失败消息发送给指定的交换机，这个交换机再发送给指定的队列，这样我们开发人员就可以从队列获取到错误信息，这样子开发人员就可以介入从而处理异常我们下面来演示将失败消息投递到指定交换机RepublishMessageRecover方案实例我们新建配置类，做好错误交换机/队列以及失败处理策略的配置然后我们再来测试错误：控制台成功显示把错误信息发送给特定交换机了来到rabbitMQ控制台，可以成功看到错误消息以及错误原因，非常方便程序员介入处理业务幂等性幂等性用在我们软件业务上也很好理解，一个业务执行多次它的结果都是一样的，那么这个业务就是幂等的，比如查询商品就是幂等的，扣减库存就不是幂等的，因为多次执行库存就会一直减少了解完幂等的概念，我们MQ对于非幂等的业务这里还会出现部分差错，比如：MQ给消费者发送“扣减库存”的消息，消费者执行完成后，返回ACK给MQ，但是此时，MQ和消费者之间的网络断开了，导致消费者的ACK没有发送给MQ，MQ就会认为消费者宕机了没有收到它发送的消息，于是又会准备发“扣减库存”的消息给消费者，当网络恢复时，MQ再次向“以为宕机”的消费者发送扣减库存的消息，消费者再次接收消息，执行相关服务，于是库存就又被扣减了对于非幂等的业务，我们有两种主要的处理方案唯一消息ID给MQ里面的消息都“设置”唯一ID，消费者接收消息后处理好业务把消息ID保存到数据库，如果下次又收到了相同消息，只需要去数据库查看该消息ID是否存在，若存在则放弃处理这个ID时MQ消息的一种属性，我们可以这样设置：在我们上一篇文章末尾介绍的消息转换器，把所发送的消息转换为MQ专属的“Message”类型对象，ID就是Message对象中的一个属性，因此我们在消息转换器中打开相关设置，然后消费者接收消息的时候去get message对象以及它的ID属性即可因此，我们在发送方的启动类修改一下我们之前定义的消息转换器，把自动创建消息id的值改为true我们设置完尝试发送消息，发现队列中的消息成功附赠id属性发送方设置完了，我们消费者接收也得接收相应类型的参数，我们前文提过的Message对象消费者也能成功接收，并且从对象中提取出相关属性业务判断这个其实就是对于特定的业务做相关的优化了，不算很泛的一种方式，这里就简单讲一下吧，实际开发中需要更具特定的业务来思考要怎么样优化场景：支付服务和交易服务，用户点击支付，交易服务吧订单标记为已支付，但是此时MQ和消费者之间的网络断开了，没有返回ACK，因此MQ又准备发送同样的“标记支付”消息。但是此时用户又想退款了，并且此时MQ和消费者之间的网络又好了，那么先前的“标记支付”的消息就会直接覆盖掉用户这次“退款消息”解决方案也很简单：消费者接收到消息去执行相关订单状态操作的业务时，去查看当前订单状态，若当前订单状态和现在一样，则不执行该业务/不管该消息，比如现在就是已支付状态，又让我干一次“已支付”那就不干了延迟消息设想一个场景：假设创建一个订单，首先通过“交易服务”创建保存订单信息，再通过“商品服务”来扣减相关商品的库存，同时调用“支付服务”来更新订单状态(是否支付)。此时如果有某一些不可抗力的因素导致“支付服务”的返回结果无法返回给交易服务，导致订单的状态一直是“未支付”，这就会导致很大的问题，假设用户已经支付了，但是又显示未支付，体验非常差。假设用户没支付，但是前面调用“商品服务”的时候，商品库存已经减少了，导致别的用户无法购买该商品，同样体验非常差因此我们可以使用延迟消息来解决该问题交易服务创建完订单后可以设置一个“闹钟”，让这个“闹钟”再设定时间段叫醒自己去查看支付服务的的返回结果，若支付服务有相关返回结果，那么就更新订单状态，如果没有返回结果，这说明“支付超时”，则取消订单，复原商品的库存延迟消息就是用来充当这个“闹钟”的，交易服务通过发送延迟消息到MQ，指定时间后发送消息给交易服务，让交易服务去查询支付服务的返回结果那我们要如何实现“延迟消息”呢，我们的rabbitMQ本身并不具备“延迟消息”的功能，但是我们可以通过里面其他属性去“模拟”延迟消息从而达到我们所想要的效果死信交换机顾名思义，就是由“死信”和“交换机”共同组成的先来了解一下什么是“死信”(如下图所示)，其中对于实现“延迟消息”最重要的就是其中的第二条：如果消息是一个过期消息，并且超时无人消费，那么该消息就会变成“死信”了解完这个特性后，我们通过死信交换机来实现延迟消息就非常容易了：我们先创建一个普通的交换机和队列，但是这个队列不绑定任何的消费者(重点!)，这样我们如果发送一些定期超时消息，由于没有人消费，留在队列里面的消息就会变成死信，然后再在该队列中绑定另一组队列和交换机(注意：这里的绑定需要使用dead-letter-exchange=绑定的交换机名称，来绑定，这样子队列中的死信就会进入所绑定的交换机)，这一组交换机和队列就会接收所绑定的队列发来的死信，最后再发送给消费者我们举例子来看，假设我们发一个30s过期的消息到第一组交换机和队列中，由于第一组交换机队列没有任何的消费者，那么该消息等到30s后会自动变为死信，而死信就会接着进入我们绑定的死信交换机队列再发送给消费者。我们去除过程只看结果，我们发送的定期过期消息不就变成了”延迟消息“了吗，定期过期多久就代表了延迟多久发送到消费者实践接下来我们来试着实践一下，其实思路很简单，无非就是创建两组队列和交换机，然后用特定的“死信”绑定方式来绑定第一组队列和第二组死信交换机即可首先先来声明死信交换机以及它的队列和消费者：再来声明绑定死信交换机的普通队列和交换机，由于普通队列不能有消费者，因此只能用bean对象的方式声明，不能像上面死信交换机一样使用注解的方式声明要注意，普通队列绑定其死信交换机所用到的key要和死信交换机绑定它的死信队列所用的key保持完全一致，这里可以看到都是“dead”发送延迟消息最后我们来看测试结果：消息在31秒的时候发出消费者在36秒的时候才接收到完全符合我们使用死信交换机实现延迟消息延迟消息插件上文讲解了死信交换实现延迟消息，但是过于麻烦你，社区中已经出现了专门设置交换机为“延迟消息交换机”的插件，从而实现延迟消息使用也非常简单，就是在交换机加一个delay属性即可声明好消费者最后编写好测试方法可以看到成功实现了延迟消息