消息中间件-消息的可靠性传递

消息中间件-消息的可靠性传递

前言

消息中间件的可靠性消息传递，是消息中间件领域很是重要的方案落实问题（在这以前的MQ理论，MQ选型是抽象层次更高的问题，这里不谈）。

而且这个问题与平常开发是存在较大的关联的。能够这么说，凡是使用了MQ的，机会都要考虑这个问题。固然也有一些原始数据采集，日志数据收集等应用场景对此没有太高要求。可是大多数的业务场景，对此仍是有着较高要求的。好比订单系统，支付系统，消息系统等，你弄丢一条消息，嘿嘿。

网上对于这方面的博客，大多从单一MQ，或者干脆就是在论述MQ。我不喜欢这样的论述，这样的论述太过局限，也过于拖沓。

此次，主要从理论方面论证消息的可靠性传递的落实。具体技术，都是依据这些理论的，具体实现都差很少。不过为了便于你们理解，我在文中会以RabbitMq，Kafka这两个主流MQ稍做举例。

在平常开发中，我更倾向于在具体开发前，先整理思路，走通理论，再开始编码。毕竟，若是连理论都走不一样，还谈什么编码。

另外，我按照消息可靠性层次逐步推动，造成相应的目录，但愿你们喜欢（由于我认为，相较网上这方面现有博客的目录，这样的目录更合理，更人性化）。

概述

这里简单谈一些有关消息可靠性传递的理论。

消息传递次数

消息在消息系统（生产者+MQ+消费者），其消费的次数，无非一下三种状况：

• 最多一次
• 最少一次
• 很少很多一次

消息可靠性层次

这也表明着消息系统的消息可靠性的三个层次：

• 最多一次：上游服务的消息发出了，至于下游能不能收到服务，就无论了。结果就是下游服务，可能根本就没有接收到消息。
• 最少一次：上游服务的消息发出了，并经过某些机制，确保下游服务必定收到了该消息。可是收到了几回，就无论了。结果就是下游服务，可能屡次收到同一条消息。
• 很少很多一次：上游服务的消息发出了，并确保下游服务必定收到了消息。下游服务经过某些机制，确保屡次收到该消息与单次收到该消息，对其系统状态的影响是相同的。

方案落实

实现上述三个层次，须要逐步从三个方面考虑：

• 最多一次：会用消息队列便可，只要确保消息的连通性便可
• 最少一次：经过MQ提供的确认机制，确保消息的传递
• 很少很多一次：经过外部应用程序，确保消息的单次消费与屡次消费对系统状态影响是一致的

上述三个层次，对系统的性能损耗，系统复杂度等都是逐步上升的。

固然，咱们首先，须要了解这三个层次分别如何实现。
再在实际开发中，根据须要，灵活选取合适方案。

最多一次的消息传递

这个方案是最简单的，只要确保消息系统的正确运做，以及系统的连通性便可。在正常状况下，能够保证绝大部分数据的可靠性传递。可是仍旧存在极小数据的丢失，而且数据的丢失会由于消息队列的选择，以及消息并发量，而受到影响。

优势

• 实现简单。只要搭建对应的MQ服务器，写出对应的生产者与消费者，以及相应配置，便可正常工做。

缺点

• 没法保证数据的可靠性，会存在必定的数据丢失状况，尤为是在并发量较大时

实际应用

能够应用于日志上传这样对消息可靠性要求低的应用场景。

总结

若是数据量不大的状况下，推荐使用RabbitMQ，其消息可靠性在地数据量下，是最可靠的。可是在达到万级并发时，会存在消息丢失，丢失的比例能够达到千分之一。

若是数据量较大的状况下，要么采用集群。要么就采用Kafk（Kafka可支持十万级并发）

通常来讲，这种消息可靠性多见于项目初建，或相似日志采集，原始数据采集这样的特定场景。

最少一次的消息传递

这个方案开始利用MQ提供的特定机制，来提升消息传递的可靠性。

优势

• 不错的消息可靠性。确保不会出现消息丢失的状况
• 实现并不复杂。只须要合理使用MQ的API，设置合理参数（如重试次数）便可

缺点

• 会出现消息重复消费的状况
• 参数的设置须要合理。如重试次数，通常设置为5次，也可根据状况，进行调整
• 资源占用的提高。如带宽（每次消息成功生产，消费都须要返回一条数据进行确认）等

方案落实

该方案的实现组成，由如下三个方面构成：

• 消息的可靠生产
• 消息的可靠存储
• 消息的可靠消费

经过以上三个方面的落实，确保可消息必定被下游服务消费。

消息的可靠生产

消息的可靠生产，是经过回调确认机制，确保消息必定被消息服务器接收。

消息生产，发送给消息服务器后，消息服务器会返回一个确认信息，表示数据正常接收。

若是生产者在必定时间内没有接收到确认信息，就会触发重试机制，进行消息的重发。

如RabbitMq的comfirm机制，Kafka的acks机制等。

RabbitMq的confirm机制存在三个模式：

• 普通模式：channel.waitForConfirms()
• 批量模式：channel.waitForConfirmsOrDie()
• 异步模式：channel.addConfirmListener()

这三个模式，看名称就能够知道具体做用了。若是但愿了解具体代码落实，详见RabbitMQ事务和Confirm发送方消息确认——深刻解读，其中确认机制写得较为简洁。

至于Kafka的acks机制，一样存在三个模式：

• acks = 0 ：不须要Kafka的任何Partition确认，即确认发送成功（这个之确保消息发送出去了，并不保证消息服务器是否成功接收）
• acks = 1 ：（默认）须要Kafka的Partition Leader确认，即被Kafka的一个Partition（Leader）接收。可是这样依旧存在极小几率的消息丢失，即Partition Leader获取了对应消息，并给了acks确认回复。可是在其余Partition同步前，Partition Leader宕机，数据丢失。那么这就形成了消息丢失。
• acks = all ：须要Kafka对应ISR中的所有Partition确认，才确认消息发送成功（固然，这里假定Kafka是多节点集群，若是只有一个分区，那就毫无心义了）。

说到这里，简单说一下，上述的操做可能形成消息的重复生产。

最简单的例子，消息成功发送，可是对应的消息确认信息因为网络波动而丢失。那么生产者就会重复发送该消息，因此消息服务器接收到了两条相同消息，故产生了消息的重复生产。

另外，上述的重试，都是存在响应时长判断（超出1min，就认为数据丢失），以及重试次数限制（超过5次，就不进行重试。不然，大量重试数据可能会拖垮整个服务）。

消息的可靠存储

消息的可靠存储，是确保消息在消息服务器通过，或者说堆积时不会由于宕机，网络等情况，丢失消息。

网上不少博客在论述消息的可靠性传递时，经常把这点遗漏。由于他们理所固然地认为消息队列已经经过集群等实现了消息队列服务的可用性，故消息的可靠性存储也就实现了。

可是这里存在两个问题。第一，可靠性不等于可用性。第二，消息的可靠存储，做为消息可靠性传递的一部分，是不可缺失的。

可用性：确保服务的可用。即对应的服务，能够提供服务。

可靠性：确保服务的正确。即对应的服务，提供的是正确的服务。

区别：我浏览淘宝，淘宝页面打不开，这就涉及了可用性问题（可用性计算公式：可用时间/所有时长*100%）。而我浏览淘宝，查询订单，给我显示的是别人的订单，这就涉及了可靠性问题。

另外这里再纠正一点，可靠性并不依赖于可用性。即便我打不开淘宝页面，我也不能说淘宝提供订单查询就有问题（只是若是没有了可用性，谈论可靠性是很是没有意义的。毕竟都用不了了，谁还关心其内容是否正确呢，都看不到）

消息队列的可用性，是经过多个节点构成集群，避免单点故障，从而提高可用性。

消息队列的可靠存储，是经过备份实现（这里不纠结备份如何确保正确）的。如RabbitMq集群的MemNode与DiskNode，又或者Kafka的replication机制等。

消息的可靠消费

消息的可靠消费，就是确保消息被消费者获取，并被成功消费。避免因为消息丢失，或者消费者宕机而形成消息消费不成功，最终形成消息的丢失（由于RabbitMq服务器在认为消息被成功消费后，将对应数据删除或标记为“已消费”）。

至于消息的可靠消费，核心理念仍是重试，重试，再重试。不过具体的实现就八仙过海，各显神通了。

这里分别说一下RabbitMq，Kafka，Rocket三者对于可靠消费的处理：

RabbitMq

提供ack机制。默认是auto，直接在拿到消息时，直接ack。确保了消息到达了消费者，可是没法解决消费者消费失败这样的问题。

实际开发中，为了确保消息的可靠消费，通常会设置为munal，只有在程序正确运行后，才会调用对应api，表示消息正确消费。

Kafka

因为Kafka的消息是落地到硬盘文件的，并且Kafka的消息分发方式是pull的，因此消息的拉取是经过offset机制去确认对应位置消息的。

固然，Kafka的offset默认是自动提交的（可经过nable_auto_commit与auto_commit_interval_ms控制）。

因此消费者调用服务失败等缘由，能够经过手动offset提交，来实现对数据的重复消费（甚至是历史数据的消费），也就能够在消费失败时对同一消息进行再消费。

若是是消费者宕机等缘由，因为Kafka服务器没有收到对应的offset提交，因此认为那条消息没有被消费成功，故返回的依旧是那条消息。

RocketMq

其实RocketMq的处理有些相似Kafka确认机制+RabbitMq死信队列的感受。

首先，消费者从RocketMq拉取消息，若是成功消费，就返回确认消息。

若是未成功消费，就尝试从新消费。

尝试消费必定次数后（如5次），就会将该消息发送之RocketMq中的重试队列。

若是遇到消费者宕机的状况，RocketMq会认为该消息未成功消费，会被其余消费者继续消费。

其实在RabbitMq的可靠性消费时，咱们也会将屡次消费失败的数据保存下来，便于后期修复等。不过保存的方式由不少种，日志，数据库，消息队列等。而RocketMq则给出了具体的落实方案。

上述的操做，可能形成消息的重复消费。

最简单的例子，消息成功被消费者消费，可是消费者还没来得及发送确认信息，就宕机了。

消息队列因为没有收到确认消息，认为该条消息还没有被消息，就将该消息交由其余消费者继续消费。

很少很多一次的消息传递

这个方案，就是经过MQ之外的应用程序，来进行扩展，最终达到消息准确消费的目的。

那么为何不将这个功能，囊括在MQ中呢？

我的认为有四个方面的考虑：

• 消息中间件，应该明确其功能域，而消息生产与消息消费每每涉及业务，因此避免与业务的耦合。因此消息中间件只完善了可靠存储。
• 准确消费，每每涉及MQ之外的部分，须要其余部分的配合。就相似与XA接口同样。这样会带来编码的约束，系统的耦合性等。
• 准确消费的实现能够经过一个工具，模块去实现，可是不应硬编码。毕竟现有的处理方案并不必定就是最优解（尤为是在调控中心，TCC框架展示的如今）。
• 性能影响。为了一个不通用的功能，会带来消息中间件的性能大幅降低

优点

• 确保消息被准确消费（很少很多一次）

缺点

• 实现复杂（生产者与消费者都须要创建对应数据库）
• 须要创建对应规范（可是通用规范肯定后，实现就会变得快速）
• 资源占用的提高。如带宽（每次消息成功生产，消费都须要返回一条数据进行确认）等

存在的问题

消息存储部分的准确存储，不应咱们来操心，因此只阐述消息生产与消息消费两个部分。

消息的重复生产

• 消息发给了消息队列服务器，消息队列服务器的确认信息因为网络波动等，没有及时到达生产者
• 消息发送给了消息队列服务器，生产者在接收消息前，宕机
• 消息发送给了消息队列服务器，生产者在接收消息后，还没来得及进行确认逻辑，宕机

综上来看，就是消息发出后，到生产者消息确认信息的处理之间，出现各类意外，致使重复生产。

消息的重复消费

• 消息已经被消费，消费者还没来得及发送确认信息，就宕机了
• 消息已经被消费，消费者发出确认信息，确认信息因为网络波动等，没有及时到达消息队列服务器
• 消息已经被消费，消费者发出确认信息，消息队列服务器对应实例在接收到确认信息前，宕机
• 消息已经被消费，消费者发出确认信息，消息队列服务器接受到了确认信息，还没来得及进行确认逻辑，宕机

综上来看，就是消息已经被消费后，到消息队列服务器进行确认消息处理之间，出现各类意外，致使重复消费。

解决方案

解决方案：messageId+幂等

准确来讲，解决方案的核心是幂等，而messageId是做为辅助手段的。

幂等

幂等，简单说明一下，就是屡次操做与单次操做对系统状态的影响是一致的。

如

i = 1;

就是幂等操做，由于不管进行几回，i的值都没有变化。

而

i++;

则不是幂等操做，由于i的值与执行次数息息相关。

故经过幂等操做来确保同一条消息，不被执行屡次。

messageId

可是，消费者如何肯定是否为同一条消息呢？

有的消息体存在惟一性字段，如orderId等。但有的消息并无这样的惟一性字段。

因此须要一个专门的字段，来表示惟一性，而且与业务消息解耦。这就是messageId。

既能够采用消息体的惟一性字段（能够是单一字段，也能够是组合字段），也能够经过特定方式生成对应标识（分布式系统，须要注意不一样实例生产者产生相同标识的可能，详见分布式全局惟一ID的实现）。

具体的生成状况，就不在这里赘述了。

方案落实

先来一张大图（这种事情，图片展现最直观了），展现一下流程：

（图片是绝对清晰的。看不清图片的朋友，请将图片在新页面打开，或下载。说实话，来到新公司，首先提高的就是画图能力。囧）

简单说一下流程，你们能够对照着上图，看一下：

生产者到消息中间件服务器

1. 生产者根据须要发送的消息，生成对应messageId。并封装对应message至生产者数据库（该操做应该利用事务性，确保生产者事件处理与message保存至数据库的原子性），同时标注message状态为sending（发送中状态）
2. 将对应message发送给消息队列服务器
3. 若是没有收到生产确认信息，则从新发送message（若是这个时候遇到生产者实例宕机，也不用担忧。由于后续会有补偿程序，进行补偿重发操做）
4. 当收到消息中间件服务器的消息生产确认消息（即肯定消息已经达到消息中间件服务器），将数据库中对应message的状态修改成sended（已发送状态）

上述中提到的补偿机制，实际上是相似事务中的一个操做。经过一个定时任务，定时巡检数据库处于sending状态的message，并经过生产者极性发送（因此message通常都保存source，target等信息）。

之因此会有sending状态的message，就是由于存在生产者消息发送出去了，还没收到生产确认信息，结果生产者实例本身宕机的状况。

至于补偿机制的定时任务，是一个很是简单的实现，这里就再也不赘述了。

消息中间件到消费者

这里进行的操做是针对非幂等的操做。

若是是幂等操做，则能够直接进行。毕竟屡次执行与单次执行对数据库的影响是一致的。

可是注意幂等操做在部分场景下无效的问题（时间影响上），如“余额 = 1k”的操做对于数据库而言是幂等的，可是在两次“余额 = 1k”操做间，有一个“余额 = 2k”的操做，则会发生问题（丢失了“余额 = 2k”操做）。固然，这种相似ABA问题，彻底能够引入版本号，来进行解决。

综上，仍是推荐采用如下解决方法，流程较为简单：

1. 消费者获取数据
2. 消费者判断数据库是否有对应message
3. 若是存在对应message，则放弃执行（由于这是一个重复操做）
4. 若是不存在，则进行相关消息处理。并经过事务控制，在消费者数据库中添加message（确保消息的处理与数据库添加message是原子操做）

至此，消息的准确传递就完成了。

总结

消息可靠性传递的发展过程，也体现了人们对消息中间件功能的一步步追求，更是体现了工程师们解决问题的思路。

不少时候，咱们会遇到不少问题，甚至使人感到杂乱不堪，无从下手。这个时候，最好的办法就是静下心来，对它们进行划分（按照重要程度，紧迫度，实现难度），再进行一个长期规划，一步步来解决。每每这个时候，动动笔，在笔记上列下清单，会是一个不错的办法。

其中消息的准确传递，涉及一些事务相关的内容。也许有人已经联想到，消息队列是否能够做为分布式事务的一种手段呢？我会在以后的博客中，来阐述分布式事务这一重要主题。

若是有什么问题或想法，能够私信或@我。

愿与诸君共进步。