rocketmq怎么保证数据不会重复_搞懂分布式技术19：使用RocketMQ事务消息解决分布式事务...

本文转载自 http://linkedkeeper.com

本文转载自 http://linkedkeeper.com

本文内容参考网络&＃xff0c;侵删

本系列文章将整理到我在GitHub上的《Java面试指南》仓库&＃xff0c;更多精彩内容请到我的仓库里查看

https://github.com/h2pl/Java-Tutorial

喜欢的话麻烦点下Star哈

文章首发于我的个人博客&＃xff1a;

http://www.how2playlife.com

该系列博文会告诉你什么是分布式系统&＃xff0c;这对后端工程师来说是很重要的一门学问&＃xff0c;我们会逐步了解常见的分布式技术、以及一些较为常见的分布式系统概念&＃xff0c;同时也需要进一步了解zookeeper、分布式事务、分布式锁、负载均衡等技术&＃xff0c;以便让你更完整地了解分布式技术的具体实战方法&＃xff0c;为真正应用分布式技术做好准备。

如果对本系列文章有什么建议&＃xff0c;或者是有什么疑问的话&＃xff0c;也可以关注公众号【Java技术江湖】联系作者&＃xff0c;欢迎你参与本系列博文的创作和修订。

初步认识RocketMQ的核心模块

rocketmq模块

rocketmq-broker&＃xff1a;接受生产者发来的消息并存储&＃xff08;通过调用rocketmq-store&＃xff09;&＃xff0c;消费者从这里取得消息。

rocketmq-client&＃xff1a;提供发送、接受消息的客户端API。

rocketmq-namesrv&＃xff1a;NameServer&＃xff0c;类似于Zookeeper&＃xff0c;这里保存着消息的TopicName&＃xff0c;队列等运行时的元信息。&＃xff08;有点NameNode的味道&＃xff09;

rocketmq-common&＃xff1a;通用的一些类&＃xff0c;方法&＃xff0c;数据结构等

rocketmq-remoting&＃xff1a;基于Netty4的client/server &＃43; fastjson序列化 &＃43; 自定义二进制协议

rocketmq-store&＃xff1a;消息、索引存储等

rocketmq-filtersrv&＃xff1a;消息过滤器Server&＃xff0c;需要注意的是&＃xff0c;要实现这种过滤&＃xff0c;需要上传代码到MQ&＃xff01;【一般而言&＃xff0c;我们利用Tag足以满足大部分的过滤需求&＃xff0c;如果更灵活更复杂的过滤需求&＃xff0c;可以考虑filtersrv组件】

rocketmq-tools&＃xff1a;命令行工具

分布式消息队列RocketMQ--事务消息--解决分布式事务

说到分布式事务&＃xff0c;就会谈到那个经典的”账号转账”问题&＃xff1a;2个账号&＃xff0c;分布处于2个不同的DB&＃xff0c;或者说2个不同的子系统里面&＃xff0c;A要扣钱&＃xff0c;B要加钱&＃xff0c;如何保证原子性&＃xff1f;

一般的思路都是通过消息中间件来实现“最终一致性”&＃xff1a;A系统扣钱&＃xff0c;然后发条消息给中间件&＃xff0c;B系统接收此消息&＃xff0c;进行加钱。

但这里面有个问题&＃xff1a;A是先update DB&＃xff0c;后发送消息呢&＃xff1f; 还是先发送消息&＃xff0c;后update DB&＃xff1f;

假设先update DB成功&＃xff0c;发送消息网络失败&＃xff0c;重发又失败&＃xff0c;怎么办&＃xff1f; 假设先发送消息成功&＃xff0c;update DB失败。消息已经发出去了&＃xff0c;又不能撤回&＃xff0c;怎么办&＃xff1f;

所以&＃xff0c;这里下个结论&＃xff1a; 只要发送消息和update DB这2个操作不是原子的&＃xff0c;无论谁先谁后&＃xff0c;都是有问题的。

那这个问题怎么解决呢&＃xff1f;

错误的方案0

有人可能想到了&＃xff0c;我可以把“发送消息”这个网络调用和update DB放在同1个事务里面&＃xff0c;如果发送消息失败&＃xff0c;update DB自动回滚。这样不就保证2个操作的原子性了吗&＃xff1f;

这个方案看似正确&＃xff0c;其实是错误的&＃xff0c;原因有2&＃xff1a;

&＃xff08;1&＃xff09;网络的2将军问题&＃xff1a;发送消息失败&＃xff0c;发送方并不知道是消息中间件真的没有收到消息呢&＃xff1f;还是消息已经收到了&＃xff0c;只是返回response的时候失败了&＃xff1f;

如果是已经收到消息了&＃xff0c;而发送端认为没有收到&＃xff0c;执行update db的回滚操作。则会导致A账号的钱没有扣&＃xff0c;B账号的钱却加了。

&＃xff08;2&＃xff09;把网络调用放在DB事务里面&＃xff0c;可能会因为网络的延时&＃xff0c;导致DB长事务。严重的&＃xff0c;会block整个DB。这个风险很大。

基于以上分析&＃xff0c;我们知道&＃xff0c;这个方案其实是错误的&＃xff01;

方案1–业务方自己实现

假设消息中间件没有提供“事务消息”功能&＃xff0c;比如你用的是Kafka。那如何解决这个问题呢&＃xff1f;

解决方案如下&＃xff1a; &＃xff08;1&＃xff09;Producer端准备1张消息表&＃xff0c;把update DB和insert message这2个操作&＃xff0c;放在一个DB事务里面。

&＃xff08;2&＃xff09;准备一个后台程序&＃xff0c;源源不断的把消息表中的message传送给消息中间件。失败了&＃xff0c;不断重试重传。允许消息重复&＃xff0c;但消息不会丢&＃xff0c;顺序也不会打乱。

&＃xff08;3&＃xff09;Consumer端准备一个判重表。处理过的消息&＃xff0c;记在判重表里面。实现业务的幂等。但这里又涉及一个原子性问题&＃xff1a;如果保证消息消费 &＃43; insert message到判重表这2个操作的原子性&＃xff1f;

消费成功&＃xff0c;但insert判重表失败&＃xff0c;怎么办&＃xff1f;关于这个&＃xff0c;在Kafka的源码分析系列&＃xff0c;第1篇&＃xff0c; exactly once问题的时候&＃xff0c;有过讨论。

通过上面3步&＃xff0c;我们基本就解决了这里update db和发送网络消息这2个操作的原子性问题。

但这个方案的一个缺点就是&＃xff1a;需要设计DB消息表&＃xff0c;同时还需要一个后台任务&＃xff0c;不断扫描本地消息。导致消息的处理和业务逻辑耦合额外增加业务方的负担。

方案2 – RocketMQ 事务消息

为了能解决该问题&＃xff0c;同时又不和业务耦合&＃xff0c;RocketMQ提出了“事务消息”的概念。

具体来说&＃xff0c;就是把消息的发送分成了2个阶段&＃xff1a;Prepare阶段和确认阶段。

具体来说&＃xff0c;上面的2个步骤&＃xff0c;被分解成3个步骤&＃xff1a; (1) 发送Prepared消息 (2) update DB (3) 根据update DB结果成功或失败&＃xff0c;Confirm或者取消Prepared消息。

可能有人会问了&＃xff0c;前2步执行成功了&＃xff0c;最后1步失败了怎么办&＃xff1f;这里就涉及到了RocketMQ的关键点&＃xff1a;RocketMQ会定期&＃xff08;默认是1分钟&＃xff09;扫描所有的Prepared消息&＃xff0c;询问发送方&＃xff0c;到底是要确认这条消息发出去&＃xff1f;还是取消此条消息&＃xff1f;

具体代码实现如下&＃xff1a;

也就是定义了一个checkListener&＃xff0c;RocketMQ会回调此Listener&＃xff0c;从而实现上面所说的方案。

然后执行本地事务&＃xff0c;具体代码如下

上面所说的消息中间件上注册的listener&＃xff0c;超时以后会回调producer的接口以确定事务执行情况

总结&＃xff1a;对比方案2和方案1&＃xff0c;RocketMQ最大的改变&＃xff0c;其实就是把“扫描消息表”这个事情&＃xff0c;不让业务方做&＃xff0c;而是消息中间件帮着做了。

至于消息表&＃xff0c;其实还是没有省掉。因为消息中间件要询问发送方&＃xff0c;事物是否执行成功&＃xff0c;还是需要一个“变相的本地消息表”&＃xff0c;记录事物执行状态。

人工介入

可能有人又要说了&＃xff0c;无论方案1&＃xff0c;还是方案2&＃xff0c;发送端把消息成功放入了队列&＃xff0c;但消费端消费失败怎么办&＃xff1f;

消费失败了&＃xff0c;重试&＃xff0c;还一直失败怎么办&＃xff1f;是不是要自动回滚整个流程&＃xff1f;

答案是人工介入。从工程实践角度讲&＃xff0c;这种整个流程自动回滚的代价是非常巨大的&＃xff0c;不但实现复杂&＃xff0c;还会引入新的问题。比如自动回滚失败&＃xff0c;又怎么处理&＃xff1f;

对应这种极低概率的case&＃xff0c;采取人工处理&＃xff0c;会比实现一个高复杂的自动化回滚系统&＃xff0c;更加可靠&＃xff0c;也更加简单。