13518219792
本文以消息中间件为例，为大家分析消息中间件的可靠性消息传递，阅读完整文相信大家对消息中间件的可靠性消息传递有了一定的认识。
消息中间件的可靠性消息传递，是消息中间件领域非常重要的方案落实问题（在这之前的MQ理论，MQ选型是抽象层次更高的问题，这里不谈）。
并且这个问题与日常开发是存在较大的关联的。可以这么说，凡是使用了MQ的，机会都要考虑这个问题。当然也有一些原始数据采集，日志数据收集等应用场景对此没有过高要求。但是大多数的业务场景，对此还是有着较高要求的。比如订单系统，支付系统，消息系统等，你弄丢一条消息，嘿嘿。
网上对于这方面的博客，大多从单一MQ，或者干脆就是在论述MQ。我不喜欢这样的论述，这样的论述太过局限，也过于拖沓。
这次，主要从理论方面论证消息的可靠性传递的落实。具体技术，都是依据这些理论的，具体实现都差不多。不过为了便于大家理解，我在文中会以RabbitMq，Kafka这两个主流MQ稍作举例。
在日常开发中，我更倾向于在具体开发前，先整理思路，走通理论，再开始编码。毕竟，如果连理论都走不同，还谈什么编码。
另外，我按照消息可靠性层次逐步推进，形成相应的目录，希望大家喜欢（因为我认为，相较网上这方面现有博客的目录，这样的目录更合理，更人性化）。
这里简单谈一些有关消息可靠性传递的理论。
消息在消息系统（生产者+MQ+消费者），其消费的次数，无非一下三种情况：
这也代表着消息系统的消息可靠性的三个层次：
实现上述三个层次，需要逐步从三个方面考虑：
上述三个层次，对系统的性能损耗，系统复杂度等都是逐步上升的。
当然，我们首先，需要了解这三个层次分别如何实现。
再在实际开发中，根据需要，灵活选取合适方案。
这个方案是最简单的，只要确保消息系统的正确运作，以及系统的连通性即可。在正常情况下，可以保证绝大部分数据的可靠性传递。但是仍旧存在极小数据的丢失，并且数据的丢失会因为消息队列的选择，以及消息并发量，而受到影响。
可以应用于日志上传这样对消息可靠性要求低的应用场景。
如果数据量不大的情况下，推荐使用RabbitMQ，其消息可靠性在地数据量下，是最可靠的。但是在达到万级并发时，会存在消息丢失，丢失的比例可以达到千分之一。
如果数据量较大的情况下，要么采用集群。要么就采用Kafk（Kafka可支持十万级并发）
一般来说，这种消息可靠性多见于项目初建，或类似日志采集，原始数据采集这样的特定场景。
这个方案开始利用MQ提供的特定机制，来提高消息传递的可靠性。
该方案的实现组成，由以下三个方面构成：
通过以上三个方面的落实，确保可消息一定被下游服务消费。
消息的可靠生产，是通过回调确认机制，确保消息一定被消息服务器接收。
消息生产，发送给消息服务器后，消息服务器会返回一个确认信息，表示数据正常接收。
如果生产者在一定时间内没有接收到确认信息，就会触发重试机制，进行消息的重发。
如RabbitMq的comfirm机制，Kafka的acks机制等。
RabbitMq的confirm机制存在三个模式：
这三个模式，看名称就可以知道具体作用了。
至于Kafka的acks机制，同样存在三个模式：
说到这里，简单说一下，上述的操作可能造成消息的重复生产。
最简单的例子，消息成功发送，但是对应的消息确认信息由于网络波动而丢失。那么生产者就会重复发送该消息，所以消息服务器接收到了两条相同消息，故产生了消息的重复生产。
另外，上述的重试，都是存在响应时长判断（超出1min，就认为数据丢失），以及重试次数限制（超过5次，就不进行重试。否则，大量重试数据可能会拖垮整个服务）。
消息的可靠存储，是确保消息在消息服务器经过，或者说堆积时不会因为宕机，网络等状况，丢失消息。
网上很多博客在论述消息的可靠性传递时，常常把这点遗漏。因为他们理所当然地认为消息队列已经通过集群等实现了消息队列服务的可用性，故消息的可靠性存储也就实现了。
但是这里存在两个问题。第一，可靠性不等于可用性。第二，消息的可靠存储，作为消息可靠性传递的一部分，是不可缺失的。
可用性：确保服务的可用。即对应的服务，可以提供服务。
可靠性：确保服务的正确。即对应的服务，提供的是正确的服务。
区别：我浏览淘宝，淘宝页面打不开，这就涉及了可用性问题（可用性计算公式：可用时间/全部时长*100%）。而我浏览淘宝，查询订单，给我显示的是别人的订单，这就涉及了可靠性问题。
另外这里再纠正一点，可靠性并不依赖于可用性。即使我打不开淘宝页面，我也不能说淘宝提供订单查询就有问题（只是如果没有了可用性，谈论可靠性是非常没有意义的。毕竟都用不了了，谁还关心其内容是否正确呢，都看不到）
消息队列的可用性，是通过多个节点构成集群，避免单点故障，从而提升可用性。
消息队列的可靠存储，是通过备份实现（这里不纠结备份如何确保正确）的。如RabbitMq集群的MemNode与DiskNode，又或者Kafka的replication机制等。
消息的可靠消费，就是确保消息被消费者获取，并被成功消费。避免由于消息丢失，或者消费者宕机而造成消息消费不成功，最终造成消息的丢失（因为RabbitMq服务器在认为消息被成功消费后，将对应数据删除或标记为“已消费”）。
至于消息的可靠消费，核心理念还是重试，重试，再重试。不过具体的实现就八仙过海，各显神通了。
这里分别说一下RabbitMq，Kafka，Rocket三者对于可靠消费的处理：
提供ack机制。默认是auto，直接在拿到消息时，直接ack。确保了消息到达了消费者，但是无法解决消费者消费失败这样的问题。
实际开发中，为了确保消息的可靠消费，一般会设置为munal，只有在程序正确运行后，才会调用对应api，表示消息正确消费。
由于Kafka的消息是落地到硬盘文件的，而且Kafka的消息分发方式是pull的，所以消息的拉取是通过offset机制去确认对应位置消息的。
当然，Kafka的offset默认是自动提交的（可通过nable_auto_commit与auto_commit_interval_ms控制）。
所以消费者调用服务失败等原因，可以通过手动offset提交，来实现对数据的重复消费（甚至是历史数据的消费），也就可以在消费失败时对同一消息进行再消费。
如果是消费者宕机等原因，由于Kafka服务器没有收到对应的offset提交，所以认为那条消息没有被消费成功，故返回的依旧是那条消息。
其实RocketMq的处理有些类似Kafka确认机制+RabbitMq死信队列的感觉。
首先，消费者从RocketMq拉取消息，如果成功消费，就返回确认消息。
如果未成功消费，就尝试重新消费。
尝试消费一定次数后（如5次），就会将该消息发送之RocketMq中的重试队列。
如果遇到消费者宕机的情况，RocketMq会认为该消息未成功消费，会被其他消费者继续消费。
其实在RabbitMq的可靠性消费时，我们也会将多次消费失败的数据保存下来，便于后期修复等。不过保存的方式由很多种，日志，数据库，消息队列等。而RocketMq则给出了具体的落实方案。
上述的操作，可能造成消息的重复消费。
最简单的例子，消息成功被消费者消费，但是消费者还没来得及发送确认信息，就宕机了。
消息队列由于没有收到确认消息，认为该条消息尚未被消息，就将该消息交由其他消费者继续消费。
这个方案，就是通过MQ以外的应用程序，来进行扩展，最终达到消息准确消费的目的。
那么为什么不将这个功能，囊括在MQ中呢？
个人认为有四个方面的考虑：
消息存储部分的准确存储，不该我们来操心，所以只阐述消息生产与消息消费两个部分。
综上来看，就是消息发出后，到生产者消息确认信息的处理之间，出现各种意外，导致重复生产。
综上来看，就是消息已经被消费后，到消息队列服务器进行确认消息处理之间，出现各种意外，导致重复消费。
解决方案：messageId+幂等
准确来说，解决方案的核心是幂等，而messageId是作为辅助手段的。
幂等，简单说明一下，就是多次操作与单次操作对系统状态的影响是一致的。
如
i = 1;就是幂等操作，因为无论进行几次，i的值都没有变化。
而
i++;则不是幂等操作，因为i的值与执行次数息息相关。
故通过幂等操作来确保同一条消息，不被执行多次。
但是，消费者如何确定是否为同一条消息呢？
有的消息体存在唯一性字段，如orderId等。但有的消息并没有这样的唯一性字段。
所以需要一个专门的字段，来表示唯一性，并且与业务消息解耦。这就是messageId。
既可以采用消息体的唯一性字段（可以是单一字段，也可以是组合字段），也可以通过特定方式生成对应标识。
具体的生成情况，就不在这里赘述了。
先来一张大图（这种事情，图片展示最直观了），展示一下流程：
（图片是绝对清晰的。看不清图片的朋友，请将图片在新页面打开，或下载。说实话，来到新公司，首先提升的就是画图能力。囧）
简单说一下流程，大家可以对照着上图，看一下：
上述中提到的补偿机制，其实是类似事务中的一个操作。通过一个定时任务，定时巡检数据库处于sending状态的message，并通过生产者极性发送（所以message一般都保存source，target等信息）。
之所以会有sending状态的message，就是因为存在生产者消息发送出去了，还没收到生产确认信息，结果生产者实例自己宕机的情况。
至于补偿机制的定时任务，是一个非常简单的实现，这里就不再赘述了。
这里进行的操作是针对非幂等的操作。
如果是幂等操作，则可以直接进行。毕竟多次执行与单次执行对数据库的影响是一致的。
但是注意幂等操作在部分场景下无效的问题（时间影响上），如“余额 = 1k”的操作对于数据库而言是幂等的，但是在两次“余额 = 1k”操作间，有一个“余额 = 2k”的操作，则会发生问题（丢失了“余额 = 2k”操作）。当然，这种类似ABA问题，完全可以引入版本号，来进行解决。
综上，还是推荐采用以下解决方法，流程较为简单：
至此，消息的准确传递就完成了。
看完上述内容，你们对消息中间件的可靠性消息传递有进一步的了解吗？如果还想学到更多技能或想了解更多相关内容，欢迎关注创新互联行业资讯频道，感谢各位的阅读。
另外有需要云服务器可以了解下创新互联scvps.cn，海内外云服务器15元起步，三天无理由+7*72小时售后在线，公司持有idc许可证，提供“云服务器、裸金属服务器、高防服务器、香港服务器、美国服务器、虚拟主机、免备案服务器”等云主机租用服务以及企业上云的综合解决方案，具有“安全稳定、简单易用、服务可用性高、性价比高”等特点与优势，专为企业上云打造定制，能够满足用户丰富、多元化的应用场景需求。
