保证MQ消息传递的一致性

我会带着你远行 2021-07-04 20:32 423阅读 0赞

[https://my.oschina.net/floor/blog/1587537][https_my.oschina.net_floor_blog_1587537]

# 1.保证消息传递与一致性 #

## 1.1生产者确保消息自主性 ##

当生产者发送一条消息时，它必须完成他的所有业务操作。

如下图：

![172415_2pMS_2296755.png][]

这保证消费者接受到消息时，生产者已处理完毕相关业务，也就是1PC的基础。

## 1.2 MQ保存并转发消息 ##

消息标记为持久化，MQ将会利用保存并转发机制，来履行它与发送者之间的契约。

至于activemq高可用部分，详见另外一篇blog：[https://my.oschina.net/floor/blog/1574213][https_my.oschina.net_floor_blog_1574213]

一般MQ保存并转发的流程如下：

![172620_KV2N_2296755.png][]

## 1.3消费者确认模式的选择 ##

### JMS提供的3中模式： ###

*  auto\_acknowledge：

消费者接收到消息后，立即自动向MQ发送确认消息。

*  dups\_ok\_acknowledge

一条消息可以被多次消费，为了确保“一次且仅仅一次发送语义”。影响MQ性能。

*  client\_acknowledge

client端确认模式，手动确认消息已消费。

### 我是如何选择的： ###

dups\_ok\_acknowledge首先被排除，是因为它影响MQ的性能，我公司使用的activemq的性能本就一般，所以没有选择。

client\_acknowledge 需要程序员编码确认消息被消费，可能存在1条消息很久没有消费掉，MQ堵住的情况，所以没有选择。

我们最终选择了auto\_acknowledge ，原因是SImple is best。至于各位如何选择，按需而定吧！

## 1.4 漏洞与补救办法 ##

经过之前的选择，我们的MQ流程图如下

![172727_OOaU_2296755.png][]

稍微看下这个流程，大家就会发现，如果消费者，在自动确认消息后，在还没有消费消息时，若消费者挂掉了 ，由于MQ在auto\_acknowledge下，当前生产者不会重新发送，这就产生了消息不一致的情况，即生产者端已处理，消费者端未处理的问题。

### 1.4.1. 2种方式处理消息不一致的情况 ###

1.生产者再发一条消息，2.消费者查询下生产者业务是否完毕。2种方式均可实现。下边主要说下我公司的方式，方案不唯一仅供参考。

#### 1.4.1.1 生产者再次发送，即定时任务补偿+幂等消费的方式 ####

这里不使用MQ的dups\_ok\_acknowledge，是因为会影响MQ的性能。生产者再次发送方式存在3个问题：

*  生产者如何知道消费者没有消费一条消息？
 *  生产者重新发送的频率是多少？
 *  消费者如何处理重复的消息？

生产者如何知道消费者没有消费一条消息？

我们公司加入了一个event表，作为生产者与消费者之间的桥梁，用来维护消息的消费状态。event表的核心字段有，JMS队列名称，业务ID（我公司发送的消息都是业务ID，这里可以是业务bean），完成状态（完成，未完成），其他字段可按需而定。

生产者重新发送的频率是多少？

需要根据业务的实时性要求和消费者的能力和可以堆积的信息进行判断。经过分析与压测后，设定定时补偿的频率。

消费者如何处理重复的消息？

为何会出现重复的消息？

举例说明：

假设正常情况下一条消息消费需要2s（即单节点一分钟消费30条），我们3分补偿一次。

消息队列中已堆积了199条消息，第200条为当前发送的消息，为了简单仅仅考虑单一消费者的情况，当3分钟后，才消费完90条消息，还有堆积110，注意这个时候消费者还没有消费生产者3分钟前发送的消息，而补偿机制又发送了一条消息进入MQ，这就出现了消费者接受到重复消息的情况。

如何解决：

我在生产中的解决办法是在：利用event表，redis实现分布式事务锁，实现幂等消费。

当消费者接受到消息后，按如下步骤处理：

1.  先查询event中的消息状态，如果消息存在且未处理，继续往下，若已处理直接返回。
2.  利用redis，加锁，可以使用redission框架，也可以利用String类型的并设置失效时间的简单方式实现不可重入的锁，个人推荐推荐redission，但是我公司使用的是String类型的并设置失效时间的简单方式。
3.  加锁成功后，再查询event中的消息状态，如果消息存在且未处理，继续往下，若已处理直接返回。
4.  消费者处理自己的业务
5.  更新event表中的状态为已处理
6.  解锁操作

熟悉并发的朋友，会发现1-3步骤是DCL模型。

综上所述，以生产者再次发送的方式，保证消费者消费消息的整体流程如下：

![174417_Emto_2296755.png][]

看到这个模型图，可能觉得比较复杂，除了第8步，我们都可以在基类中实现了，并且由于event表数据独立于MQ，我们可以做一个监控（仅仅自己考虑公司没有实现），针对event，查询消息的消费情况，还能实现人工重发功能。以上模式，已在在生产模式中大量使用。

该模式的优点是可以确保消息最终一致性，生产者，MQ，消费者压力均不大，我们公司利用该模式实现核心业务，例如 票购买后的，拆票操作，追号触发追一期操作等。

#### 1.4.1.2.消费者进行查询，推拉结合方式。 ####

按理说消费者可以启动一个定时任务，查询生产者需要它消费的数据。模型类似之前，但是我公司并没有使用这种方式。因为是不想影响消费者的消费能力。

但是我们在通知业务中，实现了一种简易的推拉结合的方式，该方式个人认为使用面比较窄，但对通知业务有一定的适用性，在这里做下简要介绍

实现方式：

仅仅提供了一个http接口供用户查询，该http接口不一定在生产者，这里仅仅是画在生产者中。

其模型图如下：

![174616_sqhL_2296755.png][]

这个模式使用的前提是：

消费者不消费数据，也对业务没有影响。

案例说明：

以通知3D的开奖号码为例：

生产者为抓取服务，当抓取服务，抓到的3D彩果后，针对每一个订阅者生存独一无二的消息数据，之后发送MQ。

消费者为push服务（实际上是调用第三方推送），接收的消息发给订阅的用户。

由于消息已入库，会在通知中心中展示，而用户是否接受到推送并不重要，他可以在app的消息中心中查询。

# 2.扩展，使用Event-Sourcing+MQ解决RPC类型的分布式事务‘ #

这个方式，来自于黄勇老师，我们目前在工作中用在用户支付与订单状态更新上，下边大部分的文字和截图的，都是来自黄勇老师的《架构探险-轻量级微服务架构下册》，在这里感谢黄勇老师的传道与解惑。

## 2.1什么是Event-Sourcing？ ##

它是一种基于事件回溯的解决方案，一般将它应用在领域对象模型中。事件不可枚举，事件类型可以枚举。我们以event表示事件表,其大体内容如下：

*  ID event的唯一表示
 *  EventType： 事件类型：CREATE，UPDATE，DELETE等。
 *  Model Name：表示模型名称：例如Foo，Bar等。
 *  MOdel ID: 模型ID
 *  Create Time 创建时间，精确到毫秒。

## 2.2实现方式： ##

### 第一步，操作模型表与时间表并写入消息队列 ###

![174816_VOLj_2296755.png][]

### 第二步，从消息队列中获取事件 ，操作模型表，若有异常，将事件再写入消息队列 ###

![174831_xp6t_2296755.png][]

### 第三步，从消息队列中获取原事件，操作事件表与模型表，进行“事件回溯” ###

![174850_absn_2296755.png][]

回溯操作的特殊说明：

*  INSERT，的逆向操作为DELETE,但是一般业务是标记删除，也就是说逆向操作为UPDATE。
 *  UPDATE，的逆向操作为UPDATE
 *  DELETE，由于一般是标记删除，所有逆向操作也是UPDATE。

## 2.3Event-Sourcing和MQ的RPC事务的控制流程图 ##

![174905_ZMV7_2296755.png][]

# 总结 #

这篇文章，是我工作中使用MQ的感悟，可能存在不对的地方，欢迎各位指正。

本文3中模式，

*  定时补偿+幂等消费
 *  推拉结合
 *  Event-Sourcing和MQ，实现RPC式分布式事务

[https_my.oschina.net_floor_blog_1587537]: https://my.oschina.net/floor/blog/1587537
[172415_2pMS_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/172415_2pMS_2296755.png
[https_my.oschina.net_floor_blog_1574213]: https://my.oschina.net/floor/blog/1574213
[172620_KV2N_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/172620_KV2N_2296755.png
[172727_OOaU_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/172727_OOaU_2296755.png
[174417_Emto_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174417_Emto_2296755.png
[174616_sqhL_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174616_sqhL_2296755.png
[174816_VOLj_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174816_VOLj_2296755.png
[174831_xp6t_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174831_xp6t_2296755.png
[174850_absn_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174850_absn_2296755.png
[174905_ZMV7_2296755.png]: https://static.oschina.net/uploads/space/2017/1210/174905_ZMV7_2296755.png